From 5809d014de3bedab7b9f533c0af1b8fc7c5dd248 Mon Sep 17 00:00:00 2001 From: Viacheslav Klimov Date: Wed, 22 May 2024 12:11:32 +0300 Subject: [PATCH] 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 --- .../server/actors/ActorSystemContext.java | 4 + .../RuleChainActorMessageProcessor.java | 3 +- .../queue/DefaultTbClusterService.java | 17 +-- .../transport/DefaultTransportApiService.java | 1 + .../controller/BaseQueueControllerTest.java | 95 +++++++++++- .../housekeeper/HousekeeperServiceTest.java | 1 - .../queue/DefaultTbClusterServiceTest.java | 3 + .../server/common/data/queue/Queue.java | 7 + .../thingsboard/server/common/msg/TbMsg.java | 61 ++++++-- common/message/src/main/proto/tbmsg.proto | 6 +- common/proto/src/main/proto/queue.proto | 2 + .../common/TbRuleEngineProducerService.java | 74 ++++++++++ .../queue/discovery/HashPartitionService.java | 137 ++++++++++++------ .../queue/discovery/PartitionService.java | 4 + .../queue/discovery/QueueRoutingInfo.java | 9 +- .../kafka/TbKafkaConsumerStatsService.java | 4 +- .../service/DefaultTransportService.java | 48 +----- .../queue/tenant-profile-queues.component.ts | 3 +- .../profile/tenant-profile.component.ts | 9 +- .../queue/queue-form.component.html | 25 ++-- .../components/queue/queue-form.component.ts | 7 +- ui-ngx/src/app/shared/models/queue.models.ts | 15 +- .../assets/locale/locale.constant-en_US.json | 1 + 23 files changed, 397 insertions(+), 139 deletions(-) create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/common/TbRuleEngineProducerService.java diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index f411344e3f..b125daa591 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/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(); } diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java index a16e6122f5..eaa2218116 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java +++ b/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 ruleNodeRelations = nodeRoutes.get(originatorNodeId); if (ruleNodeRelations == null) { // When unchecked, this will cause NullPointerException when rule node doesn't exist anymore diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java index 7af2579b89..bce5dece3b 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java +++ b/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()); diff --git a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java index 6dcf84e721..f5bd42e29d 100644 --- a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java +++ b/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()); } diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseQueueControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseQueueControllerTest.java index b0a6db944c..ad52934587 100644 --- a/application/src/test/java/org/thingsboard/server/controller/BaseQueueControllerTest.java +++ b/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 msgCaptor = ArgumentCaptor.forClass(TbActorMsg.class); + verify(actorSystemContext, atLeastOnce()).tell(msgCaptor.capture()); + List 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); + } + } diff --git a/application/src/test/java/org/thingsboard/server/service/housekeeper/HousekeeperServiceTest.java b/application/src/test/java/org/thingsboard/server/service/housekeeper/HousekeeperServiceTest.java index e8f1a43ad1..06740e2851 100644 --- a/application/src/test/java/org/thingsboard/server/service/housekeeper/HousekeeperServiceTest.java +++ b/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" diff --git a/application/src/test/java/org/thingsboard/server/service/queue/DefaultTbClusterServiceTest.java b/application/src/test/java/org/thingsboard/server/service/queue/DefaultTbClusterServiceTest.java index 33b0fc7801..22f3620d8c 100644 --- a/application/src/test/java/org/thingsboard/server/service/queue/DefaultTbClusterServiceTest.java +++ b/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; diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/queue/Queue.java b/common/data/src/main/java/org/thingsboard/server/common/data/queue/Queue.java index 1c08072e61..ad5b9cf808 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/queue/Queue.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/queue/Queue.java @@ -73,4 +73,11 @@ public class Queue extends BaseDataWithAdditionalInfo 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); + } + } diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java index cbf9ae34b3..47fb80c90e 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java +++ b/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() { diff --git a/common/message/src/main/proto/tbmsg.proto b/common/message/src/main/proto/tbmsg.proto index 4c2d110888..fc9265aa14 100644 --- a/common/message/src/main/proto/tbmsg.proto +++ b/common/message/src/main/proto/tbmsg.proto @@ -66,4 +66,8 @@ message TbMsgProto { int64 customerIdLSB = 18; TbMsgProcessingCtxProto ctx = 19; -} \ No newline at end of file + + int64 correlationIdMSB = 20; + int64 correlationIdLSB = 21; + int32 partition = 22; +} diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 53a3026424..faacf05968 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/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 { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/TbRuleEngineProducerService.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/TbRuleEngineProducerService.java new file mode 100644 index 0000000000..49b40e3a6d --- /dev/null +++ b/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> producer, + TenantId tenantId, TbMsg tbMsg, TbQueueCallback callback) { + List 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> 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); + } + +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java index e5ad2a3622..04d55f91c0 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java +++ b/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 partitionTopicsMap = new ConcurrentHashMap<>(); private final ConcurrentMap partitionSizesMap = new ConcurrentHashMap<>(); + private final ConcurrentMap queueConfigs = new ConcurrentHashMap<>(); private final ConcurrentMap 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 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 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 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 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 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 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(); + } + } + } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java index e14a6c3947..b5744981bd 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java +++ b/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 resolveAll(ServiceType serviceType, String queueName, TenantId tenantId, EntityId entityId); + boolean isMyPartition(ServiceType serviceType, TenantId tenantId, EntityId entityId); List getMyPartitions(QueueKey queueKey); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/QueueRoutingInfo.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/QueueRoutingInfo.java index 51976d89f3..ba56044bb4 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/QueueRoutingInfo.java +++ b/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(); } + } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatsService.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatsService.java index 56440cf4a5..29b8064f04 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatsService.java +++ b/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); } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index 9312370c34..c0c7339eba 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/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 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 { diff --git a/ui-ngx/src/app/modules/home/components/profile/queue/tenant-profile-queues.component.ts b/ui-ngx/src/app/modules/home/components/profile/queue/tenant-profile-queues.component.ts index bee66eef53..a6eda87d4a 100644 --- a/ui-ngx/src/app/modules/home/components/profile/queue/tenant-profile-queues.component.ts +++ b/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); diff --git a/ui-ngx/src/app/modules/home/components/profile/tenant-profile.component.ts b/ui-ngx/src/app/modules/home/components/profile/tenant-profile.component.ts index 5e7ab536a9..9035b3dbaa 100644 --- a/ui-ngx/src/app/modules/home/components/profile/tenant-profile.component.ts +++ b/ui-ngx/src/app/modules/home/components/profile/tenant-profile.component.ts @@ -75,7 +75,8 @@ export class TenantProfileComponent extends EntityComponent { topic: 'tb_rule_engine.main', additionalInfo: { description: '', - customProperties: '' + customProperties: '', + duplicateMsgToAllPartitions: false } }, { @@ -99,7 +100,8 @@ export class TenantProfileComponent extends EntityComponent { }, additionalInfo: { description: '', - customProperties: '' + customProperties: '', + duplicateMsgToAllPartitions: false } }, { @@ -123,7 +125,8 @@ export class TenantProfileComponent extends EntityComponent { }, additionalInfo: { description: '', - customProperties: '' + customProperties: '', + duplicateMsgToAllPartitions: false } } ]; diff --git a/ui-ngx/src/app/modules/home/components/queue/queue-form.component.html b/ui-ngx/src/app/modules/home/components/queue/queue-form.component.html index 7039cbda23..38e2860cd5 100644 --- a/ui-ngx/src/app/modules/home/components/queue/queue-form.component.html +++ b/ui-ngx/src/app/modules/home/components/queue/queue-form.component.html @@ -207,14 +207,19 @@ - - queue.custom-properties - - queue.custom-properties-hint - - - queue.description - - queue.description-hint - +
+ + {{ 'queue.duplicate-msg-to-all-partitions' | translate }} + + + queue.custom-properties + + queue.custom-properties-hint + + + queue.description + + queue.description-hint + +
diff --git a/ui-ngx/src/app/modules/home/components/queue/queue-form.component.ts b/ui-ngx/src/app/modules/home/components/queue/queue-form.component.ts index f54d99e332..c9a09a262d 100644 --- a/ui-ngx/src/app/modules/home/components/queue/queue-form.component.ts +++ b/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(); diff --git a/ui-ngx/src/app/shared/models/queue.models.ts b/ui-ngx/src/app/shared/models/queue.models.ts index 1b750f6880..eb23cc1269 100644 --- a/ui-ngx/src/app/shared/models/queue.models.ts +++ b/ui-ngx/src/app/shared/models/queue.models.ts @@ -108,20 +108,21 @@ export interface QueueInfo extends BaseData, 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; }; } diff --git a/ui-ngx/src/assets/locale/locale.constant-en_US.json b/ui-ngx/src/assets/locale/locale.constant-en_US.json index fcb0021ddc..885aaeaefe 100644 --- a/ui-ngx/src/assets/locale/locale.constant-en_US.json +++ b/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)",