Queue option to duplicate msgs to all partitions (#10815)

* Queue config option to duplicate msgs to all partitions

* UI for queue config option to duplicate msgs to all partitions

* Fixes for msg duplication; add tests

* Introduce TbRuleEngineProducerService

* Use TbRuleEngineProducerService in TbClusterService

* Fix DefaultTbClusterServiceTest

* Add correlationId to TbMsg

* Refactor TbMsg

* Support msg duplication for all msg types

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

@@ -636,6 +636,10 @@ public class ActorSystemContext {
         return partitionService.resolve(serviceType, queueName, tenantId, entityId);
     }
 
+    public TopicPartitionInfo resolve(TenantId tenantId, EntityId entityId, TbMsg msg) {
+        return partitionService.resolve(ServiceType.TB_RULE_ENGINE, msg.getQueueName(), tenantId, entityId, msg.getPartition());
+    }
+
     public String getServiceId() {
         return serviceInfoProvider.getServiceId();
     }

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

@@ -42,7 +42,6 @@ import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
 import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
 import org.thingsboard.server.common.msg.queue.RuleEngineException;
 import org.thingsboard.server.common.msg.queue.RuleNodeException;
-import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.common.stats.TbApiUsageReportClient;
 import org.thingsboard.server.dao.rule.RuleChainService;
@@ -295,7 +294,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
         try {
             checkComponentStateActive(msg);
             EntityId entityId = msg.getOriginator();
-            TopicPartitionInfo tpi = systemContext.resolve(ServiceType.TB_RULE_ENGINE, msg.getQueueName(), tenantId, entityId);
+            TopicPartitionInfo tpi = systemContext.resolve(tenantId, entityId, msg);
 
             List<RuleNodeRelation> ruleNodeRelations = nodeRoutes.get(originatorNodeId);
             if (ruleNodeRelations == null) { // When unchecked, this will cause NullPointerException when rule node doesn't exist anymore

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

@@ -30,7 +30,6 @@ import org.thingsboard.server.common.data.EdgeUtils;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.HasName;
 import org.thingsboard.server.common.data.HasRuleEngineProfile;
-import org.thingsboard.server.common.data.TbResource;
 import org.thingsboard.server.common.data.TbResourceInfo;
 import org.thingsboard.server.common.data.Tenant;
 import org.thingsboard.server.common.data.TenantProfile;
@@ -59,8 +58,8 @@ import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse;
 import org.thingsboard.server.common.msg.rule.engine.DeviceEdgeUpdateMsg;
 import org.thingsboard.server.common.msg.rule.engine.DeviceNameOrTypeUpdateMsg;
-import org.thingsboard.server.dao.edge.EdgeService;
 import org.thingsboard.server.common.util.ProtoUtils;
+import org.thingsboard.server.dao.edge.EdgeService;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.FromDeviceRPCResponseProto;
 import org.thingsboard.server.gen.transport.TransportProtos.QueueDeleteMsg;
@@ -74,6 +73,7 @@ import org.thingsboard.server.queue.TbQueueCallback;
 import org.thingsboard.server.queue.TbQueueProducer;
 import org.thingsboard.server.queue.common.MultipleTbQueueCallbackWrapper;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
+import org.thingsboard.server.queue.common.TbRuleEngineProducerService;
 import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.discovery.TopicService;
 import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
@@ -116,6 +116,9 @@ public class DefaultTbClusterService implements TbClusterService {
     @Lazy
     private TbQueueProducerProvider producerProvider;
 
+    @Autowired
+    private TbRuleEngineProducerService ruleEngineProducerService;
+
     @Autowired
     @Lazy
     private OtaPackageStateService otaPackageStateService;
@@ -203,13 +206,8 @@ public class DefaultTbClusterService implements TbClusterService {
             HasRuleEngineProfile ruleEngineProfile = getRuleEngineProfileForEntityOrElseNull(tenantId, entityId);
             tbMsg = transformMsg(tbMsg, ruleEngineProfile);
         }
-        TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueName(), tenantId, entityId);
-        log.trace("PUSHING msg: {} to:{}", tbMsg, tpi);
-        ToRuleEngineMsg msg = ToRuleEngineMsg.newBuilder()
-                .setTenantIdMSB(tenantId.getId().getMostSignificantBits())
-                .setTenantIdLSB(tenantId.getId().getLeastSignificantBits())
-                .setTbMsg(TbMsg.toByteString(tbMsg)).build();
-        producerProvider.getRuleEngineMsgProducer().send(tpi, new TbProtoQueueMsg<>(tbMsg.getId(), msg), callback);
+
+        ruleEngineProducerService.sendToRuleEngine(producerProvider.getRuleEngineMsgProducer(), tenantId, tbMsg, callback);
         toRuleEngineMsgs.incrementAndGet();
     }
 
@@ -584,6 +582,7 @@ public class DefaultTbClusterService implements TbClusterService {
                         .setQueueName(queue.getName())
                         .setQueueTopic(queue.getTopic())
                         .setPartitions(queue.getPartitions())
+                        .setDuplicateMsgToAllPartitions(queue.isDuplicateMsgToAllPartitions())
                         .build())
                 .collect(Collectors.toList());
 

application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java

@@ -675,6 +675,7 @@ public class DefaultTransportApiService implements TransportApiService {
                                 .setQueueName(queue.getName())
                                 .setQueueTopic(queue.getTopic())
                                 .setPartitions(queue.getPartitions())
+                                .setDuplicateMsgToAllPartitions(queue.isDuplicateMsgToAllPartitions())
                                 .build()).collect(Collectors.toList())).build());
     }
 

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

@@ -16,6 +16,8 @@
 package org.thingsboard.server.controller;
 
 import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.Assert;
 import org.junit.Test;
@@ -25,9 +27,13 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.test.context.TestPropertySource;
 import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.common.data.Device;
+import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.exception.ThingsboardException;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.data.msg.TbMsgType;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.common.data.queue.ProcessingStrategy;
@@ -36,14 +42,19 @@ import org.thingsboard.server.common.data.queue.Queue;
 import org.thingsboard.server.common.data.queue.QueueStats;
 import org.thingsboard.server.common.data.queue.SubmitStrategy;
 import org.thingsboard.server.common.data.queue.SubmitStrategyType;
+import org.thingsboard.server.common.msg.TbActorMsg;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
 import org.thingsboard.server.common.msg.queue.RuleEngineException;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.stats.StatsFactory;
 import org.thingsboard.server.dao.queue.QueueStatsService;
 import org.thingsboard.server.dao.service.DaoSqlTest;
 import org.thingsboard.server.dao.timeseries.TimeseriesDao;
+import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
+import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.discovery.QueueKey;
 import org.thingsboard.server.service.queue.TbRuleEngineConsumerStats;
 import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingResult;
@@ -57,20 +68,25 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 
 @DaoSqlTest
 @TestPropertySource(properties = {
-        "queue.rule-engine.stats.max-error-message-length=100"
+        "queue.rule-engine.stats.max-error-message-length=100",
+        "transport.http.enabled=true"
 })
 public class BaseQueueControllerTest extends AbstractControllerTest {
 
@@ -82,6 +98,12 @@ public class BaseQueueControllerTest extends AbstractControllerTest {
     private TimeseriesDao timeseriesDao;
     @Autowired
     private QueueStatsService queueStatsService;
+    @SpyBean
+    private PartitionService partitionService;
+    @SpyBean
+    private TimeseriesService timeseriesService;
+    @SpyBean
+    private ActorSystemContext actorSystemContext;
 
     @Test
     public void testQueueWithServiceTypeRE() throws Exception {
@@ -236,4 +258,75 @@ public class BaseQueueControllerTest extends AbstractControllerTest {
         assertThat(finalErrorMessage).isEqualTo(largeExceptionMessage.substring(0, 100) + "...[truncated 50 symbols]");
     }
 
+    @Test
+    public void testMsgDuplicationToAllPartitions_fromTransport() throws Exception {
+        loginSysAdmin();
+        Queue queue = new Queue();
+        queue.setName("RealTime");
+        queue.setTopic("tb_rule_engine.real_time");
+        queue.setPollInterval(25);
+        int partitions = 12;
+        queue.setPartitions(partitions);
+        queue.setTenantId(TenantId.SYS_TENANT_ID);
+        queue.setConsumerPerPartition(true);
+        queue.setPackProcessingTimeout(2000);
+        SubmitStrategy submitStrategy = new SubmitStrategy();
+        submitStrategy.setType(SubmitStrategyType.BURST);
+        queue.setSubmitStrategy(submitStrategy);
+        ProcessingStrategy processingStrategy = new ProcessingStrategy();
+        processingStrategy.setType(ProcessingStrategyType.RETRY_ALL);
+        processingStrategy.setRetries(0);
+        processingStrategy.setPauseBetweenRetries(3);
+        processingStrategy.setMaxPauseBetweenRetries(5);
+        queue.setProcessingStrategy(processingStrategy);
+        queue.setAdditionalInfo(JacksonUtil.newObjectNode()
+                .put("duplicateMsgToAllPartitions", true));
+        queue = saveQueue(queue);
+
+        await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> {
+            assertThat(partitionService.resolveAll(ServiceType.TB_RULE_ENGINE, "RealTime", tenantId, tenantId)).hasSize(partitions);
+        });
+
+        loginTenantAdmin();
+        DeviceProfile deviceProfile = createDeviceProfile("realtime");
+        deviceProfile.setDefaultQueueName(queue.getName());
+        deviceProfile = doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class);
+        Device device = createDevice("test", deviceProfile.getName(), "test-token");
+
+        JsonNode payload = JacksonUtil.newObjectNode()
+                .put("test", "test");
+        doPost("/api/v1/test-token/telemetry", payload).andExpect(status().isOk());
+
+        await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> {
+            verify(timeseriesService, times(partitions)).save(eq(tenantId), eq(device.getId()),
+                    argThat(ts -> ts.size() == 1 && ts.get(0).getKey().equals("test")), anyLong());
+
+            ArgumentCaptor<TbActorMsg> msgCaptor = ArgumentCaptor.forClass(TbActorMsg.class);
+            verify(actorSystemContext, atLeastOnce()).tell(msgCaptor.capture());
+            List<TbMsg> tbMsgs = msgCaptor.getAllValues().stream()
+                    .map(actorMsg -> actorMsg instanceof QueueToRuleEngineMsg queueToRuleEngineMsg ? queueToRuleEngineMsg.getMsg() : null)
+                    .filter(tbMsg -> tbMsg != null && tbMsg.getCorrelationId() != null && tbMsg.getInternalType() == TbMsgType.POST_TELEMETRY_REQUEST)
+                    .toList();
+            assertThat(tbMsgs).hasSize(partitions);
+            UUID correlationId = tbMsgs.get(0).getCorrelationId();
+            assertThat(tbMsgs).extracting(TbMsg::getCorrelationId).containsOnly(correlationId);
+            assertThat(tbMsgs).extracting(TbMsg::getId).doesNotHaveDuplicates();
+            assertThat(tbMsgs).extracting(TbMsg::getPartition).containsExactlyInAnyOrder(IntStream.range(0, partitions).boxed().toArray(Integer[]::new));
+        });
+
+        loginSysAdmin();
+        ((ObjectNode) queue.getAdditionalInfo()).put("duplicateMsgToAllPartitions", false);
+        queue = saveQueue(queue);
+
+        await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> {
+            assertThat(partitionService.resolveAll(ServiceType.TB_RULE_ENGINE, "RealTime", tenantId, tenantId)).hasSize(1);
+        });
+
+        doDelete("/api/queues/" + queue.getUuidId()).andExpect(status().isOk());
+    }
+
+    private Queue saveQueue(Queue queue) {
+        return doPost("/api/queues?serviceType=TB_RULE_ENGINE", queue, Queue.class);
+    }
+
 }

application/src/test/java/org/thingsboard/server/service/housekeeper/HousekeeperServiceTest.java

@@ -101,7 +101,6 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.
 
 @DaoSqlTest
 @TestPropertySource(properties = {
-        "transport.http.enabled=true",
         "queue.core.housekeeper.task-reprocessing-delay-ms=2000",
         "queue.core.housekeeper.poll-interval-ms=1000",
         "queue.core.housekeeper.max-reprocessing-attempts=5"

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

@@ -33,6 +33,7 @@ import org.thingsboard.server.dao.edge.EdgeService;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.queue.TbQueueProducer;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
+import org.thingsboard.server.queue.common.TbRuleEngineProducerService;
 import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.discovery.TopicService;
 import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
@@ -77,6 +78,8 @@ public class DefaultTbClusterServiceTest {
     protected PartitionService partitionService;
     @MockBean
     protected TbQueueProducerProvider producerProvider;
+    @MockBean
+    protected TbRuleEngineProducerService ruleEngineProducerService;
 
     @SpyBean
     protected TopicService topicService;

common/data/src/main/java/org/thingsboard/server/common/data/queue/Queue.java

@@ -73,4 +73,11 @@ public class Queue extends BaseDataWithAdditionalInfo<QueueId> implements HasNam
                 .filter(JsonNode::isTextual).map(JsonNode::asText).orElse(null);
     }
 
+    @JsonIgnore
+    public boolean isDuplicateMsgToAllPartitions() {
+        return Optional.ofNullable(getAdditionalInfo())
+                .map(info -> info.get("duplicateMsgToAllPartitions"))
+                .filter(JsonNode::isBoolean).map(JsonNode::asBoolean).orElse(false);
+    }
+
 }

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

@@ -19,6 +19,8 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.InvalidProtocolBufferException;
 import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
 import lombok.Data;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
@@ -42,6 +44,8 @@ import java.util.UUID;
  */
 @Data
 @Slf4j
+@AllArgsConstructor(access = AccessLevel.PRIVATE)
+@Builder(toBuilder = true)
 public final class TbMsg implements Serializable {
 
     public static final String EMPTY_JSON_OBJECT = "{}";
@@ -60,6 +64,10 @@ public final class TbMsg implements Serializable {
     private final String data;
     private final RuleChainId ruleChainId;
     private final RuleNodeId ruleNodeId;
+
+    private final UUID correlationId;
+    private final Integer partition;
+
     @Getter(value = AccessLevel.NONE)
     @JsonIgnore
     //This field is not serialized because we use queues and there is no need to do it
@@ -257,7 +265,7 @@ public final class TbMsg implements Serializable {
     @Deprecated(since = "3.6.0")
     public static TbMsg transformMsg(TbMsg tbMsg, String type, EntityId originator, TbMsgMetaData metaData, String data) {
         return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, null, type, originator, tbMsg.customerId, metaData.copy(), tbMsg.dataType,
-                data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.callback);
+                data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.correlationId, tbMsg.partition, tbMsg.ctx.copy(), tbMsg.callback);
     }
 
     public static TbMsg newMsg(TbMsgType type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
@@ -271,54 +279,54 @@ public final class TbMsg implements Serializable {
     }
 
     public static TbMsg transformMsg(TbMsg tbMsg, TbMsgType type, EntityId originator, TbMsgMetaData metaData, String data) {
-        return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, type, originator, tbMsg.customerId, metaData.copy(), tbMsg.dataType,
-                data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.callback);
+        return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, type, type.name(), originator, tbMsg.customerId, metaData.copy(), tbMsg.dataType,
+                data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.correlationId, tbMsg.partition, tbMsg.ctx.copy(), tbMsg.callback);
     }
 
     public static TbMsg transformMsgOriginator(TbMsg tbMsg, EntityId originatorId) {
         return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, tbMsg.internalType, tbMsg.type, originatorId, tbMsg.getCustomerId(), tbMsg.metaData, tbMsg.dataType,
-                tbMsg.data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.getCallback());
+                tbMsg.data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.correlationId, tbMsg.partition, tbMsg.ctx.copy(), tbMsg.getCallback());
     }
 
     public static TbMsg transformMsgData(TbMsg tbMsg, String data) {
         return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, tbMsg.internalType, tbMsg.type, tbMsg.originator, tbMsg.customerId, tbMsg.metaData, tbMsg.dataType,
-                data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.getCallback());
+                data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.correlationId, tbMsg.partition, tbMsg.ctx.copy(), tbMsg.getCallback());
     }
 
     public static TbMsg transformMsgMetadata(TbMsg tbMsg, TbMsgMetaData metadata) {
         return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, tbMsg.internalType, tbMsg.type, tbMsg.originator, tbMsg.customerId, metadata.copy(), tbMsg.dataType,
-                tbMsg.data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.getCallback());
+                tbMsg.data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.correlationId, tbMsg.partition, tbMsg.ctx.copy(), tbMsg.getCallback());
     }
 
     public static TbMsg transformMsg(TbMsg tbMsg, TbMsgMetaData metadata, String data) {
         return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, tbMsg.internalType, tbMsg.type, tbMsg.originator, tbMsg.customerId, metadata, tbMsg.dataType,
-                data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.getCallback());
+                data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.correlationId, tbMsg.partition, tbMsg.ctx.copy(), tbMsg.getCallback());
     }
 
     public static TbMsg transformMsgCustomerId(TbMsg tbMsg, CustomerId customerId) {
         return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, tbMsg.internalType, tbMsg.type, tbMsg.originator, customerId, tbMsg.metaData, tbMsg.dataType,
-                tbMsg.data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.getCallback());
+                tbMsg.data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.correlationId, tbMsg.partition, tbMsg.ctx.copy(), tbMsg.getCallback());
     }
 
     public static TbMsg transformMsgRuleChainId(TbMsg tbMsg, RuleChainId ruleChainId) {
         return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, tbMsg.internalType, tbMsg.type, tbMsg.originator, tbMsg.customerId, tbMsg.metaData, tbMsg.dataType,
-                tbMsg.data, ruleChainId, null, tbMsg.ctx.copy(), tbMsg.getCallback());
+                tbMsg.data, ruleChainId, null, tbMsg.correlationId, tbMsg.partition, tbMsg.ctx.copy(), tbMsg.getCallback());
     }
 
     public static TbMsg transformMsgQueueName(TbMsg tbMsg, String queueName) {
         return new TbMsg(queueName, tbMsg.id, tbMsg.ts, tbMsg.internalType, tbMsg.type, tbMsg.originator, tbMsg.customerId, tbMsg.metaData, tbMsg.dataType,
-                tbMsg.data, tbMsg.getRuleChainId(), null, tbMsg.ctx.copy(), tbMsg.getCallback());
+                tbMsg.data, tbMsg.getRuleChainId(), null, tbMsg.correlationId, tbMsg.partition, tbMsg.ctx.copy(), tbMsg.getCallback());
     }
 
     public static TbMsg transformMsg(TbMsg tbMsg, RuleChainId ruleChainId, String queueName) {
         return new TbMsg(queueName, tbMsg.id, tbMsg.ts, tbMsg.internalType, tbMsg.type, tbMsg.originator, tbMsg.customerId, tbMsg.metaData, tbMsg.dataType,
-                tbMsg.data, ruleChainId, null, tbMsg.ctx.copy(), tbMsg.getCallback());
+                tbMsg.data, ruleChainId, null, tbMsg.correlationId, tbMsg.partition, tbMsg.ctx.copy(), tbMsg.getCallback());
     }
 
     //used for enqueueForTellNext
     public static TbMsg newMsg(TbMsg tbMsg, String queueName, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
         return new TbMsg(queueName, UUID.randomUUID(), tbMsg.getTs(), tbMsg.getInternalType(), tbMsg.getType(), tbMsg.getOriginator(), tbMsg.customerId, tbMsg.getMetaData().copy(),
-                tbMsg.getDataType(), tbMsg.getData(), ruleChainId, ruleNodeId, tbMsg.ctx.copy(), TbMsgCallback.EMPTY);
+                tbMsg.getDataType(), tbMsg.getData(), ruleChainId, ruleNodeId, tbMsg.correlationId, tbMsg.partition, tbMsg.ctx.copy(), TbMsgCallback.EMPTY);
     }
 
     private TbMsg(String queueName, UUID id, long ts, TbMsgType internalType, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, TbMsgDataType dataType, String data,
@@ -328,6 +336,11 @@ public final class TbMsg implements Serializable {
 
     private TbMsg(String queueName, UUID id, long ts, TbMsgType internalType, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, TbMsgDataType dataType, String data,
                   RuleChainId ruleChainId, RuleNodeId ruleNodeId, TbMsgProcessingCtx ctx, TbMsgCallback callback) {
+        this(queueName, id, ts, internalType, type, originator, customerId, metaData, dataType, data, ruleChainId, ruleNodeId, null, null, ctx, callback);
+    }
+
+    private TbMsg(String queueName, UUID id, long ts, TbMsgType internalType, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, TbMsgDataType dataType, String data,
+                  RuleChainId ruleChainId, RuleNodeId ruleNodeId, UUID correlationId, Integer partition, TbMsgProcessingCtx ctx, TbMsgCallback callback) {
         this.id = id;
         this.queueName = queueName;
         if (ts > 0) {
@@ -352,6 +365,8 @@ public final class TbMsg implements Serializable {
         this.data = data;
         this.ruleChainId = ruleChainId;
         this.ruleNodeId = ruleNodeId;
+        this.correlationId = correlationId;
+        this.partition = partition;
         this.ctx = ctx != null ? ctx : new TbMsgProcessingCtx();
         this.callback = Objects.requireNonNullElse(callback, TbMsgCallback.EMPTY);
     }
@@ -391,6 +406,14 @@ public final class TbMsg implements Serializable {
         builder.setDataType(msg.getDataType().ordinal());
         builder.setData(msg.getData());
 
+        if (msg.getCorrelationId() != null) {
+            builder.setCorrelationIdMSB(msg.getCorrelationId().getMostSignificantBits());
+            builder.setCorrelationIdLSB(msg.getCorrelationId().getLeastSignificantBits());
+        }
+        if (msg.getPartition() != null) {
+            builder.setPartition(msg.getPartition());
+        }
+
         builder.setCtx(msg.ctx.toProto());
         return builder.build().toByteArray();
     }
@@ -403,6 +426,8 @@ public final class TbMsg implements Serializable {
             CustomerId customerId = null;
             RuleChainId ruleChainId = null;
             RuleNodeId ruleNodeId = null;
+            UUID correlationId = null;
+            Integer partition = null;
             if (proto.getCustomerIdMSB() != 0L && proto.getCustomerIdLSB() != 0L) {
                 customerId = new CustomerId(new UUID(proto.getCustomerIdMSB(), proto.getCustomerIdLSB()));
             }
@@ -412,6 +437,10 @@ public final class TbMsg implements Serializable {
             if (proto.getRuleNodeIdMSB() != 0L && proto.getRuleNodeIdLSB() != 0L) {
                 ruleNodeId = new RuleNodeId(new UUID(proto.getRuleNodeIdMSB(), proto.getRuleNodeIdLSB()));
             }
+            if (proto.getCorrelationIdMSB() != 0L && proto.getCorrelationIdLSB() != 0L) {
+                correlationId = new UUID(proto.getCorrelationIdMSB(), proto.getCorrelationIdLSB());
+                partition = proto.getPartition();
+            }
 
             TbMsgProcessingCtx ctx;
             if (proto.hasCtx()) {
@@ -423,7 +452,7 @@ public final class TbMsg implements Serializable {
 
             TbMsgDataType dataType = TbMsgDataType.values()[proto.getDataType()];
             return new TbMsg(queueName, UUID.fromString(proto.getId()), proto.getTs(), null, proto.getType(), entityId, customerId,
-                    metaData, dataType, proto.getData(), ruleChainId, ruleNodeId, ctx, callback);
+                    metaData, dataType, proto.getData(), ruleChainId, ruleNodeId, correlationId, partition, ctx, callback);
         } catch (InvalidProtocolBufferException e) {
             throw new IllegalStateException("Could not parse protobuf for TbMsg", e);
         }
@@ -435,17 +464,17 @@ public final class TbMsg implements Serializable {
 
     public TbMsg copyWithRuleChainId(RuleChainId ruleChainId, UUID msgId) {
         return new TbMsg(this.queueName, msgId, this.ts, this.internalType, this.type, this.originator, this.customerId,
-                this.metaData, this.dataType, this.data, ruleChainId, null, this.ctx, callback);
+                this.metaData, this.dataType, this.data, ruleChainId, null, this.correlationId, this.partition, this.ctx, callback);
     }
 
     public TbMsg copyWithRuleNodeId(RuleChainId ruleChainId, RuleNodeId ruleNodeId, UUID msgId) {
         return new TbMsg(this.queueName, msgId, this.ts, this.internalType, this.type, this.originator, this.customerId,
-                this.metaData, this.dataType, this.data, ruleChainId, ruleNodeId, this.ctx, callback);
+                this.metaData, this.dataType, this.data, ruleChainId, ruleNodeId, this.correlationId, this.partition, this.ctx, callback);
     }
 
     public TbMsg copyWithNewCtx() {
         return new TbMsg(this.queueName, this.id, this.ts, this.internalType, this.type, this.originator, this.customerId,
-                this.metaData, this.dataType, this.data, ruleChainId, ruleNodeId, this.ctx.copy(), TbMsgCallback.EMPTY);
+                this.metaData, this.dataType, this.data, ruleChainId, ruleNodeId, this.correlationId, this.partition, this.ctx.copy(), TbMsgCallback.EMPTY);
     }
 
     public TbMsgCallback getCallback() {

common/message/src/main/proto/tbmsg.proto

@@ -66,4 +66,8 @@ message TbMsgProto {
     int64 customerIdLSB = 18;
 
     TbMsgProcessingCtxProto ctx = 19;
-}
+
+    int64 correlationIdMSB = 20;
+    int64 correlationIdLSB = 21;
+    int32 partition = 22;
+}

common/proto/src/main/proto/queue.proto

@@ -419,6 +419,7 @@ message GetQueueRoutingInfoResponseMsg {
   string queueName = 5;
   string queueTopic = 6;
   int32 partitions = 7;
+  optional bool duplicateMsgToAllPartitions = 8;
 }
 
 message QueueUpdateMsg {
@@ -429,6 +430,7 @@ message QueueUpdateMsg {
   string queueName = 5;
   string queueTopic = 6;
   int32 partitions = 7;
+  optional bool duplicateMsgToAllPartitions = 8;
 }
 
 message QueueDeleteMsg {

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

@@ -0,0 +1,74 @@
+/**
+ * Copyright © 2016-2024 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.common;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.queue.ServiceType;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
+import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
+import org.thingsboard.server.queue.TbQueueCallback;
+import org.thingsboard.server.queue.TbQueueProducer;
+import org.thingsboard.server.queue.discovery.PartitionService;
+
+import java.util.List;
+import java.util.UUID;
+
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class TbRuleEngineProducerService {
+
+    private final PartitionService partitionService;
+
+    public void sendToRuleEngine(TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> producer,
+                                 TenantId tenantId, TbMsg tbMsg, TbQueueCallback callback) {
+        List<TopicPartitionInfo> tpis = partitionService.resolveAll(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueName(), tenantId, tbMsg.getOriginator());
+        if (tpis.size() > 1) {
+            UUID correlationId = UUID.randomUUID();
+            for (int i = 0; i < tpis.size(); i++) {
+                TopicPartitionInfo tpi = tpis.get(i);
+                Integer partition = tpi.getPartition().orElse(null);
+                UUID id = i > 0 ? UUID.randomUUID() : tbMsg.getId();
+
+                tbMsg = tbMsg.toBuilder()
+                        .id(id)
+                        .correlationId(correlationId)
+                        .partition(partition)
+                        .build();
+                sendToRuleEngine(producer, tpi, tenantId, tbMsg, i == tpis.size() - 1 ? callback : null);
+            }
+        } else {
+            sendToRuleEngine(producer, tpis.get(0), tenantId, tbMsg, callback);
+        }
+    }
+
+    private void sendToRuleEngine(TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> producer, TopicPartitionInfo tpi,
+                                  TenantId tenantId, TbMsg tbMsg, TbQueueCallback callback) {
+        if (log.isTraceEnabled()) {
+            log.trace("[{}][{}] Pushing to topic {} message {}", tenantId, tbMsg.getOriginator(), tpi.getFullTopicName(), tbMsg);
+        }
+        ToRuleEngineMsg msg = ToRuleEngineMsg.newBuilder()
+                .setTbMsg(TbMsg.toByteString(tbMsg))
+                .setTenantIdMSB(tenantId.getId().getMostSignificantBits())
+                .setTenantIdLSB(tenantId.getId().getLeastSignificantBits()).build();
+        producer.send(tpi, new TbProtoQueueMsg<>(tbMsg.getId(), msg), callback);
+    }
+
+}

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

@@ -17,6 +17,7 @@ package org.thingsboard.server.queue.discovery;
 
 import com.google.common.hash.HashFunction;
 import com.google.common.hash.Hashing;
+import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.ApplicationEventPublisher;
@@ -77,6 +78,7 @@ public class HashPartitionService implements PartitionService {
 
     private final ConcurrentMap<QueueKey, String> partitionTopicsMap = new ConcurrentHashMap<>();
     private final ConcurrentMap<QueueKey, Integer> partitionSizesMap = new ConcurrentHashMap<>();
+    private final ConcurrentMap<QueueKey, QueueConfig> queueConfigs = new ConcurrentHashMap<>();
 
     private final ConcurrentMap<TenantId, TenantRoutingInfo> tenantRoutingInfoMap = new ConcurrentHashMap<>();
 
@@ -132,6 +134,7 @@ public class HashPartitionService implements PartitionService {
             QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, queue);
             partitionTopicsMap.put(queueKey, queue.getQueueTopic());
             partitionSizesMap.put(queueKey, queue.getPartitions());
+            queueConfigs.put(queueKey, new QueueConfig(queue));
         });
     }
 
@@ -174,10 +177,12 @@ public class HashPartitionService implements PartitionService {
     @Override
     public void updateQueues(List<TransportProtos.QueueUpdateMsg> queueUpdateMsgs) {
         for (TransportProtos.QueueUpdateMsg queueUpdateMsg : queueUpdateMsgs) {
-            TenantId tenantId = TenantId.fromUUID(new UUID(queueUpdateMsg.getTenantIdMSB(), queueUpdateMsg.getTenantIdLSB()));
-            QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, queueUpdateMsg.getQueueName(), tenantId);
-            partitionTopicsMap.put(queueKey, queueUpdateMsg.getQueueTopic());
-            partitionSizesMap.put(queueKey, queueUpdateMsg.getPartitions());
+            QueueRoutingInfo queueRoutingInfo = new QueueRoutingInfo(queueUpdateMsg);
+            TenantId tenantId = queueRoutingInfo.getTenantId();
+            QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, queueRoutingInfo.getQueueName(), tenantId);
+            partitionTopicsMap.put(queueKey, queueRoutingInfo.getQueueTopic());
+            partitionSizesMap.put(queueKey, queueRoutingInfo.getPartitions());
+            queueConfigs.put(queueKey, new QueueConfig(queueRoutingInfo));
             if (!tenantId.isSysTenantId()) {
                 tenantRoutingInfoMap.remove(tenantId);
             }
@@ -193,9 +198,7 @@ public class HashPartitionService implements PartitionService {
                 })
                 .collect(Collectors.toList());
         queueKeys.forEach(queueKey -> {
-            myPartitions.remove(queueKey);
-            partitionTopicsMap.remove(queueKey);
-            partitionSizesMap.remove(queueKey);
+            removeQueue(queueKey);
             evictTenantInfo(queueKey.getTenantId());
         });
         if (serviceInfoProvider.isService(ServiceType.TB_RULE_ENGINE)) {
@@ -209,14 +212,17 @@ public class HashPartitionService implements PartitionService {
         List<QueueKey> queueKeys = partitionSizesMap.keySet().stream()
                 .filter(queueKey -> tenantId.equals(queueKey.getTenantId()))
                 .collect(Collectors.toList());
-        queueKeys.forEach(queueKey -> {
-            myPartitions.remove(queueKey);
-            partitionTopicsMap.remove(queueKey);
-            partitionSizesMap.remove(queueKey);
-        });
+        queueKeys.forEach(this::removeQueue);
         evictTenantInfo(tenantId);
     }
 
+    private void removeQueue(QueueKey queueKey) {
+        myPartitions.remove(queueKey);
+        partitionTopicsMap.remove(queueKey);
+        partitionSizesMap.remove(queueKey);
+        queueConfigs.remove(queueKey);
+    }
+
     @Override
     public boolean isManagedByCurrentService(TenantId tenantId) {
         if (serviceInfoProvider.isService(ServiceType.TB_CORE) || !serviceInfoProvider.isService(ServiceType.TB_RULE_ENGINE)) {
@@ -254,6 +260,64 @@ public class HashPartitionService implements PartitionService {
 
     @Override
     public TopicPartitionInfo resolve(ServiceType serviceType, String queueName, TenantId tenantId, EntityId entityId) {
+        QueueKey queueKey = getQueueKey(serviceType, queueName, tenantId);
+        return resolve(queueKey, entityId);
+    }
+
+    @Override
+    public TopicPartitionInfo resolve(ServiceType serviceType, String queueName, TenantId tenantId, EntityId entityId, Integer partition) {
+        QueueKey queueKey = getQueueKey(serviceType, queueName, tenantId);
+        if (partition != null) {
+            return buildTopicPartitionInfo(queueKey, partition);
+        } else {
+            return resolve(queueKey, entityId);
+        }
+    }
+
+    @Override
+    public TopicPartitionInfo resolve(ServiceType serviceType, TenantId tenantId, EntityId entityId) {
+        return resolve(serviceType, null, tenantId, entityId);
+    }
+
+    @Override
+    public List<TopicPartitionInfo> resolveAll(ServiceType serviceType, String queueName, TenantId tenantId, EntityId entityId) {
+        QueueKey queueKey = getQueueKey(serviceType, queueName, tenantId);
+        TopicPartitionInfo tpi = resolve(queueKey, entityId);
+        if (serviceType != ServiceType.TB_RULE_ENGINE || tpi.getPartition().isEmpty()) {
+            return List.of(tpi);
+        }
+
+        QueueConfig queueConfig = queueConfigs.get(queueKey);
+        if (queueConfig != null && queueConfig.isDuplicateMsgToAllPartitions()) {
+            int partition = tpi.getPartition().get();
+            Integer partitionsCount = partitionSizesMap.get(queueKey);
+
+            List<TopicPartitionInfo> partitions = new ArrayList<>(partitionsCount);
+            partitions.add(tpi);
+            for (int i = 0; i < partitionsCount; i++) {
+                if (i != partition) {
+                    partitions.add(buildTopicPartitionInfo(queueKey, i, false));
+                }
+            }
+            return partitions;
+        } else {
+            return Collections.singletonList(tpi);
+        }
+    }
+
+    private TopicPartitionInfo resolve(QueueKey queueKey, EntityId entityId) {
+        Integer partitionSize = partitionSizesMap.get(queueKey);
+        if (partitionSize == null) {
+            throw new IllegalStateException("Partitions info for queue " + queueKey + " is missing");
+        }
+
+        int hash = hash(entityId.getId());
+        int partition = Math.abs(hash % partitionSize);
+
+        return buildTopicPartitionInfo(queueKey, partition);
+    }
+
+    private QueueKey getQueueKey(ServiceType serviceType, String queueName, TenantId tenantId) {
         TenantId isolatedOrSystemTenantId = getIsolatedOrSystemTenantId(serviceType, tenantId);
         if (queueName == null) {
             queueName = MAIN_QUEUE_NAME;
@@ -270,12 +334,7 @@ public class HashPartitionService implements PartitionService {
                 log.warn("Using queue {} instead of isolated {} for tenant {}", queueKey, queueName, isolatedOrSystemTenantId);
             }
         }
-        return resolve(queueKey, entityId);
-    }
-
-    @Override
-    public TopicPartitionInfo resolve(ServiceType serviceType, TenantId tenantId, EntityId entityId) {
-        return resolve(serviceType, null, tenantId, entityId);
+        return queueKey;
     }
 
     @Override
@@ -288,18 +347,6 @@ public class HashPartitionService implements PartitionService {
         }
     }
 
-    private TopicPartitionInfo resolve(QueueKey queueKey, EntityId entityId) {
-        Integer partitionSize = partitionSizesMap.get(queueKey);
-        if (partitionSize == null) {
-            throw new IllegalStateException("Partitions info for queue " + queueKey + " is missing");
-        }
-
-        int hash = hash(entityId.getId());
-        int partition = Math.abs(hash % partitionSize);
-
-        return buildTopicPartitionInfo(queueKey, partition);
-    }
-
     @Override
     public synchronized void recalculatePartitions(ServiceInfo currentService, List<ServiceInfo> otherServices) {
         log.info("Recalculating partitions");
@@ -478,18 +525,17 @@ public class HashPartitionService implements PartitionService {
     }
 
     private TopicPartitionInfo buildTopicPartitionInfo(QueueKey queueKey, int partition) {
-        TopicPartitionInfo.TopicPartitionInfoBuilder tpi = TopicPartitionInfo.builder();
-        tpi.topic(topicService.buildTopicName(partitionTopicsMap.get(queueKey)));
-        tpi.partition(partition);
-        tpi.tenantId(queueKey.getTenantId());
-
         List<Integer> partitions = myPartitions.get(queueKey);
-        if (partitions != null) {
-            tpi.myPartition(partitions.contains(partition));
-        } else {
-            tpi.myPartition(false);
-        }
-        return tpi.build();
+        return buildTopicPartitionInfo(queueKey, partition, partitions != null && partitions.contains(partition));
+    }
+
+    private TopicPartitionInfo buildTopicPartitionInfo(QueueKey queueKey, int partition, boolean myPartition) {
+        return TopicPartitionInfo.builder()
+                .topic(topicService.buildTopicName(partitionTopicsMap.get(queueKey)))
+                .partition(partition)
+                .tenantId(queueKey.getTenantId())
+                .myPartition(myPartition)
+                .build();
     }
 
     private boolean isIsolated(ServiceType serviceType, TenantId tenantId) {
@@ -619,4 +665,13 @@ public class HashPartitionService implements PartitionService {
         return serviceInfos.stream().map(ServiceInfo::getServiceId).collect(Collectors.toList());
     }
 
+    @Data
+    public static class QueueConfig {
+        private boolean duplicateMsgToAllPartitions;
+
+        public QueueConfig(QueueRoutingInfo queueRoutingInfo) {
+            this.duplicateMsgToAllPartitions = queueRoutingInfo.isDuplicateMsgToAllPartitions();
+        }
+    }
+
 }

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

@@ -33,8 +33,12 @@ public interface PartitionService {
 
     TopicPartitionInfo resolve(ServiceType serviceType, String queueName, TenantId tenantId, EntityId entityId);
 
+    TopicPartitionInfo resolve(ServiceType serviceType, String queueName, TenantId tenantId, EntityId entityId, Integer partition);
+
     TopicPartitionInfo resolve(ServiceType serviceType, TenantId tenantId, EntityId entityId);
 
+    List<TopicPartitionInfo> resolveAll(ServiceType serviceType, String queueName, TenantId tenantId, EntityId entityId);
+
     boolean isMyPartition(ServiceType serviceType, TenantId tenantId, EntityId entityId);
 
     List<Integer> getMyPartitions(QueueKey queueKey);

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

@@ -33,6 +33,7 @@ public class QueueRoutingInfo {
     private final String queueName;
     private final String queueTopic;
     private final int partitions;
+    private final boolean duplicateMsgToAllPartitions;
 
     public QueueRoutingInfo(Queue queue) {
         this.tenantId = queue.getTenantId();
@@ -40,21 +41,25 @@ public class QueueRoutingInfo {
         this.queueName = queue.getName();
         this.queueTopic = queue.getTopic();
         this.partitions = queue.getPartitions();
+        this.duplicateMsgToAllPartitions = queue.isDuplicateMsgToAllPartitions();
     }
 
     public QueueRoutingInfo(GetQueueRoutingInfoResponseMsg routingInfo) {
-        this.tenantId = new TenantId(new UUID(routingInfo.getTenantIdMSB(), routingInfo.getTenantIdLSB()));
+        this.tenantId = TenantId.fromUUID(new UUID(routingInfo.getTenantIdMSB(), routingInfo.getTenantIdLSB()));
         this.queueId = new QueueId(new UUID(routingInfo.getQueueIdMSB(), routingInfo.getQueueIdLSB()));
         this.queueName = routingInfo.getQueueName();
         this.queueTopic = routingInfo.getQueueTopic();
         this.partitions = routingInfo.getPartitions();
+        this.duplicateMsgToAllPartitions = routingInfo.hasDuplicateMsgToAllPartitions() && routingInfo.getDuplicateMsgToAllPartitions();
     }
 
     public QueueRoutingInfo(QueueUpdateMsg queueUpdateMsg) {
-        this.tenantId = new TenantId(new UUID(queueUpdateMsg.getTenantIdMSB(), queueUpdateMsg.getTenantIdLSB()));
+        this.tenantId = TenantId.fromUUID(new UUID(queueUpdateMsg.getTenantIdMSB(), queueUpdateMsg.getTenantIdLSB()));
         this.queueId = new QueueId(new UUID(queueUpdateMsg.getQueueIdMSB(), queueUpdateMsg.getQueueIdLSB()));
         this.queueName = queueUpdateMsg.getQueueName();
         this.queueTopic = queueUpdateMsg.getQueueTopic();
         this.partitions = queueUpdateMsg.getPartitions();
+        this.duplicateMsgToAllPartitions = queueUpdateMsg.hasDuplicateMsgToAllPartitions() && queueUpdateMsg.getDuplicateMsgToAllPartitions();
     }
+
 }

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

@@ -112,8 +112,8 @@ public class TbKafkaConsumerStatsService {
     }
 
     private boolean isStatsPrintRequired() {
-        boolean isMyRuleEnginePartition = partitionService.resolve(ServiceType.TB_RULE_ENGINE, TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID).isMyPartition();
-        boolean isMyCorePartition = partitionService.resolve(ServiceType.TB_CORE, TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID).isMyPartition();
+        boolean isMyRuleEnginePartition = partitionService.isMyPartition(ServiceType.TB_RULE_ENGINE, TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID);
+        boolean isMyCorePartition = partitionService.isMyPartition(ServiceType.TB_CORE, TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID);
         return log.isInfoEnabled() && (isMyRuleEnginePartition || isMyCorePartition);
     }
 

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java

@@ -23,6 +23,7 @@ import com.google.gson.Gson;
 import com.google.gson.JsonObject;
 import jakarta.annotation.PostConstruct;
 import jakarta.annotation.PreDestroy;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -99,6 +100,7 @@ import org.thingsboard.server.queue.TbQueueProducer;
 import org.thingsboard.server.queue.TbQueueRequestTemplate;
 import org.thingsboard.server.queue.common.AsyncCallbackTemplate;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
+import org.thingsboard.server.queue.common.TbRuleEngineProducerService;
 import org.thingsboard.server.queue.common.consumer.QueueConsumerManager;
 import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
@@ -131,6 +133,7 @@ import java.util.stream.Collectors;
 @Slf4j
 @Service
 @TbTransportComponent
+@RequiredArgsConstructor
 public class DefaultTransportService extends TransportActivityManager implements TransportService {
 
     public static final TransportProtos.SessionEventMsg SESSION_EVENT_MSG_OPEN = TransportProtos.SessionEventMsg.newBuilder()
@@ -163,6 +166,7 @@ public class DefaultTransportService extends TransportActivityManager implements
     private final PartitionService partitionService;
     private final TbTransportQueueFactory queueProvider;
     private final TbQueueProducerProvider producerProvider;
+    private final TbRuleEngineProducerService ruleEngineProducerService;
 
     private final TopicService topicService;
     private final TbServiceInfoProvider serviceInfoProvider;
@@ -191,34 +195,6 @@ public class DefaultTransportService extends TransportActivityManager implements
 
     private final Map<String, RpcRequestMetadata> toServerRpcPendingMap = new ConcurrentHashMap<>();
 
-    public DefaultTransportService(PartitionService partitionService,
-                                   TbServiceInfoProvider serviceInfoProvider,
-                                   TbTransportQueueFactory queueProvider,
-                                   TbQueueProducerProvider producerProvider,
-                                   TopicService topicService,
-                                   StatsFactory statsFactory,
-                                   TransportDeviceProfileCache deviceProfileCache,
-                                   TransportTenantProfileCache tenantProfileCache,
-                                   TransportRateLimitService rateLimitService,
-                                   SchedulerComponent scheduler, TransportResourceCache transportResourceCache,
-                                   ApplicationEventPublisher eventPublisher, NotificationRuleProcessor notificationRuleProcessor,
-                                   EntityLimitsCache entityLimitsCache) {
-        this.partitionService = partitionService;
-        this.serviceInfoProvider = serviceInfoProvider;
-        this.queueProvider = queueProvider;
-        this.producerProvider = producerProvider;
-        this.topicService = topicService;
-        this.statsFactory = statsFactory;
-        this.deviceProfileCache = deviceProfileCache;
-        this.tenantProfileCache = tenantProfileCache;
-        this.rateLimitService = rateLimitService;
-        this.scheduler = scheduler;
-        this.transportResourceCache = transportResourceCache;
-        this.eventPublisher = eventPublisher;
-        this.notificationRuleProcessor = notificationRuleProcessor;
-        this.entityLimitsCache = entityLimitsCache;
-    }
-
     @PostConstruct
     public void init() {
         super.init();
@@ -1109,19 +1085,6 @@ public class DefaultTransportService extends TransportActivityManager implements
         tbCoreMsgProducer.send(tpi, new TbProtoQueueMsg<>(routingKey, msg), wrappedCallback);
     }
 
-    private void sendToRuleEngine(TenantId tenantId, TbMsg tbMsg, TbQueueCallback callback) {
-        TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueName(), tenantId, tbMsg.getOriginator());
-        if (log.isTraceEnabled()) {
-            log.trace("[{}][{}] Pushing to topic {} message {}", tenantId, tbMsg.getOriginator(), tpi.getFullTopicName(), tbMsg);
-        }
-        ToRuleEngineMsg msg = ToRuleEngineMsg.newBuilder().setTbMsg(TbMsg.toByteString(tbMsg))
-                .setTenantIdMSB(tenantId.getId().getMostSignificantBits())
-                .setTenantIdLSB(tenantId.getId().getLeastSignificantBits()).build();
-        ruleEngineProducerStats.incrementTotal();
-        StatsCallback wrappedCallback = new StatsCallback(callback, ruleEngineProducerStats);
-        ruleEngineMsgProducer.send(tpi, new TbProtoQueueMsg<>(tbMsg.getId(), msg), wrappedCallback);
-    }
-
     private void sendToRuleEngine(TenantId tenantId, DeviceId deviceId, CustomerId customerId, TransportProtos.SessionInfoProto sessionInfo, JsonObject json,
                                   TbMsgMetaData metaData, TbMsgType tbMsgType, TbQueueCallback callback) {
         DeviceProfileId deviceProfileId = new DeviceProfileId(new UUID(sessionInfo.getDeviceProfileIdMSB(), sessionInfo.getDeviceProfileIdLSB()));
@@ -1139,7 +1102,8 @@ public class DefaultTransportService extends TransportActivityManager implements
         }
 
         TbMsg tbMsg = TbMsg.newMsg(queueName, tbMsgType, deviceId, customerId, metaData, gson.toJson(json), ruleChainId, null);
-        sendToRuleEngine(tenantId, tbMsg, callback);
+        ruleEngineProducerService.sendToRuleEngine(ruleEngineMsgProducer, tenantId, tbMsg, new StatsCallback(callback, ruleEngineProducerStats));
+        ruleEngineProducerStats.incrementTotal();
     }
 
     private class TransportTbQueueCallback implements TbQueueCallback {

ui-ngx/src/app/modules/home/components/profile/queue/tenant-profile-queues.component.ts

@@ -174,7 +174,8 @@ export class TenantProfileQueuesComponent implements ControlValueAccessor, Valid
       topic: '',
       additionalInfo: {
         description: '',
-        customProperties: ''
+        customProperties: '',
+        duplicateMsgToAllPartitions: false
       }
     };
     this.idMap.push(queue.id);

ui-ngx/src/app/modules/home/components/profile/tenant-profile.component.ts

@@ -75,7 +75,8 @@ export class TenantProfileComponent extends EntityComponent<TenantProfile> {
         topic: 'tb_rule_engine.main',
         additionalInfo: {
           description: '',
-          customProperties: ''
+          customProperties: '',
+          duplicateMsgToAllPartitions: false
         }
       },
       {
@@ -99,7 +100,8 @@ export class TenantProfileComponent extends EntityComponent<TenantProfile> {
         },
         additionalInfo: {
           description: '',
-          customProperties: ''
+          customProperties: '',
+          duplicateMsgToAllPartitions: false
         }
       },
       {
@@ -123,7 +125,8 @@ export class TenantProfileComponent extends EntityComponent<TenantProfile> {
         },
         additionalInfo: {
           description: '',
-          customProperties: ''
+          customProperties: '',
+          duplicateMsgToAllPartitions: false
         }
       }
     ];

ui-ngx/src/app/modules/home/components/queue/queue-form.component.html

@@ -207,14 +207,19 @@
       </ng-template>
     </mat-expansion-panel>
   </mat-accordion>
-  <mat-form-field class="mat-block" formGroupName="additionalInfo" subscriptSizing="dynamic">
-    <mat-label translate>queue.custom-properties</mat-label>
-    <textarea matInput formControlName="customProperties" cdkTextareaAutosize cdkAutosizeMinRows="1"></textarea>
-    <mat-hint translate>queue.custom-properties-hint</mat-hint>
-  </mat-form-field>
-  <mat-form-field class="mat-block" formGroupName="additionalInfo" appearance="fill">
-    <mat-label translate>queue.description</mat-label>
-    <textarea matInput formControlName="description" rows="2"></textarea>
-    <mat-hint translate>queue.description-hint</mat-hint>
-  </mat-form-field>
+  <div formGroupName="additionalInfo">
+    <mat-checkbox fxFlex formControlName="duplicateMsgToAllPartitions" style="padding-bottom: 16px;">
+      {{ 'queue.duplicate-msg-to-all-partitions' | translate }}
+    </mat-checkbox>
+    <mat-form-field class="mat-block" subscriptSizing="dynamic">
+      <mat-label translate>queue.custom-properties</mat-label>
+      <textarea matInput formControlName="customProperties" cdkTextareaAutosize cdkAutosizeMinRows="1"></textarea>
+      <mat-hint translate>queue.custom-properties-hint</mat-hint>
+    </mat-form-field>
+    <mat-form-field class="mat-block" appearance="fill">
+      <mat-label translate>queue.description</mat-label>
+      <textarea matInput formControlName="description" rows="2"></textarea>
+      <mat-hint translate>queue.description-hint</mat-hint>
+    </mat-form-field>
+  </div>
 </form>

ui-ngx/src/app/modules/home/components/queue/queue-form.component.ts

@@ -118,7 +118,8 @@ export class QueueFormComponent implements ControlValueAccessor, OnInit, OnDestr
         topic: [''],
         additionalInfo: this.fb.group({
           description: [''],
-          customProperties: ['']
+          customProperties: [''],
+          duplicateMsgToAllPartitions: [false]
         })
       });
     this.valueChange$ = this.queueFormGroup.valueChanges.subscribe(() => {
@@ -160,6 +161,10 @@ export class QueueFormComponent implements ControlValueAccessor, OnInit, OnDestr
       this.queueFormGroup.patchValue(this.modelValue, {emitEvent: false});
       this.queueFormGroup.get('additionalInfo').get('description')
         .patchValue(this.modelValue.additionalInfo?.description, {emitEvent: false});
+      this.queueFormGroup.get('additionalInfo').get('customProperties')
+        .patchValue(this.modelValue.additionalInfo?.customProperties, {emitEvent: false});
+      this.queueFormGroup.get('additionalInfo').get('duplicateMsgToAllPartitions')
+        .patchValue(this.modelValue.additionalInfo?.duplicateMsgToAllPartitions, {emitEvent: false});
       this.submitStrategyTypeChanged();
       if (!this.disabled && !this.queueFormGroup.valid) {
         this.updateModel();

ui-ngx/src/app/shared/models/queue.models.ts

@@ -108,20 +108,21 @@ export interface QueueInfo extends BaseData<QueueId>, HasTenantId {
   consumerPerPartition: boolean;
   pollInterval: number;
   processingStrategy: {
-    type: QueueProcessingStrategyTypes,
-    retries: number,
-    failurePercentage: number,
-    pauseBetweenRetries: number,
-    maxPauseBetweenRetries: number
+    type: QueueProcessingStrategyTypes;
+    retries: number;
+    failurePercentage: number;
+    pauseBetweenRetries: number;
+    maxPauseBetweenRetries: number;
   };
   submitStrategy: {
-    type: QueueSubmitStrategyTypes,
-    batchSize: number,
+    type: QueueSubmitStrategyTypes;
+    batchSize: number;
   };
   tenantId?: TenantId;
   topic: string;
   additionalInfo: {
     description?: string;
     customProperties?: string;
+    duplicateMsgToAllPartitions?: boolean;
   };
 }

ui-ngx/src/assets/locale/locale.constant-en_US.json

@@ -4277,6 +4277,7 @@
         "immediate-processing": "Immediate processing",
         "consumer-per-partition": "Send message poll for each consumer",
         "consumer-per-partition-hint": "Enable separate consumer(s) per each partition",
+        "duplicate-msg-to-all-partitions": "Duplicate message to all partitions",
         "processing-timeout": "Processing within, ms",
         "batch-size": "Batch size",
         "retries": "Number of retries (0 – unlimited)",