From 827f63ea9ff22783f7753078a0cc200abece1eab Mon Sep 17 00:00:00 2001 From: Dmytro Skarzhynets Date: Tue, 28 Oct 2025 15:32:49 +0200 Subject: [PATCH 1/3] Add tests for rule engine consumer loop --- ...faultTbCalculatedFieldConsumerService.java | 6 +- .../queue/DefaultTbCoreConsumerService.java | 6 +- .../queue/DefaultTbEdgeConsumerService.java | 6 +- .../DefaultTbRuleEngineConsumerService.java | 6 +- .../TbMsgPackProcessingContextFactory.java | 35 +++ .../processing/AbstractConsumerService.java | 6 +- .../AbstractTbRuleEngineSubmitStrategy.java | 12 +- .../BatchTbRuleEngineSubmitStrategy.java | 2 +- .../BurstTbRuleEngineSubmitStrategy.java | 2 +- .../service/queue/processing/IdMsgPair.java | 14 +- ...lByEntityIdTbRuleEngineSubmitStrategy.java | 10 +- .../SequentialTbRuleEngineSubmitStrategy.java | 6 +- .../TbRuleEngineQueueConsumerManager.java | 8 +- .../src/main/resources/thingsboard.yml | 3 + .../queue/TbMsgPackProcessingContextTest.java | 260 ++++++++++++++-- .../RuleEngineConsumerLoopTest.java | 281 ++++++++++++++++++ 16 files changed, 595 insertions(+), 68 deletions(-) create mode 100644 application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextFactory.java create mode 100644 application/src/test/java/org/thingsboard/server/service/queue/ruleengine/RuleEngineConsumerLoopTest.java diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java index acb36449e8..91981cf86f 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java @@ -146,15 +146,15 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractPartitionBa private void processMsgs(List> msgs, TbQueueConsumer> consumer, Object consumerKey, QueueConfig config) throws Exception { List> orderedMsgList = msgs.stream().map(msg -> new IdMsgPair<>(UUID.randomUUID(), msg)).toList(); ConcurrentMap> pendingMap = orderedMsgList.stream().collect( - Collectors.toConcurrentMap(IdMsgPair::getUuid, IdMsgPair::getMsg)); + Collectors.toConcurrentMap(IdMsgPair::uuid, IdMsgPair::msg)); CountDownLatch processingTimeoutLatch = new CountDownLatch(1); TbPackProcessingContext> ctx = new TbPackProcessingContext<>( processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>()); PendingMsgHolder pendingMsgHolder = new PendingMsgHolder<>(); Future packSubmitFuture = consumersExecutor.submit(() -> { orderedMsgList.forEach((element) -> { - UUID id = element.getUuid(); - TbProtoQueueMsg msg = element.getMsg(); + UUID id = element.uuid(); + TbProtoQueueMsg msg = element.msg(); log.trace("[{}] Creating main callback for message: {}", id, msg.getValue()); TbCallback callback = new TbPackCallback<>(id, ctx); try { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index 9ab5a062eb..57a7018310 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -260,15 +260,15 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService> msgs, TbQueueConsumer> consumer, Object consumerKey, QueueConfig config) throws Exception { List> orderedMsgList = msgs.stream().map(msg -> new IdMsgPair<>(UUID.randomUUID(), msg)).toList(); ConcurrentMap> pendingMap = orderedMsgList.stream().collect( - Collectors.toConcurrentMap(IdMsgPair::getUuid, IdMsgPair::getMsg)); + Collectors.toConcurrentMap(IdMsgPair::uuid, IdMsgPair::msg)); CountDownLatch processingTimeoutLatch = new CountDownLatch(1); TbPackProcessingContext> ctx = new TbPackProcessingContext<>( processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>()); PendingMsgHolder pendingMsgHolder = new PendingMsgHolder<>(); Future packSubmitFuture = consumersExecutor.submit(() -> { orderedMsgList.forEach((element) -> { - UUID id = element.getUuid(); - TbProtoQueueMsg msg = element.getMsg(); + UUID id = element.uuid(); + TbProtoQueueMsg msg = element.msg(); log.trace("[{}] Creating main callback for message: {}", id, msg.getValue()); TbCallback callback = new TbPackCallback<>(id, ctx); try { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java index 40ef9bfeca..71dacdbdaf 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java @@ -129,15 +129,15 @@ public class DefaultTbEdgeConsumerService extends AbstractConsumerService> msgs, TbQueueConsumer> consumer, Object consumerKey, QueueConfig edgeQueueConfig) throws InterruptedException { List> orderedMsgList = msgs.stream().map(msg -> new IdMsgPair<>(UUID.randomUUID(), msg)).toList(); ConcurrentMap> pendingMap = orderedMsgList.stream().collect( - Collectors.toConcurrentMap(IdMsgPair::getUuid, IdMsgPair::getMsg)); + Collectors.toConcurrentMap(IdMsgPair::uuid, IdMsgPair::msg)); CountDownLatch processingTimeoutLatch = new CountDownLatch(1); TbPackProcessingContext> ctx = new TbPackProcessingContext<>( processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>()); PendingMsgHolder pendingMsgHolder = new PendingMsgHolder<>(); Future submitFuture = consumersExecutor.submit(() -> { orderedMsgList.forEach((element) -> { - UUID id = element.getUuid(); - TbProtoQueueMsg msg = element.getMsg(); + UUID id = element.uuid(); + TbProtoQueueMsg msg = element.msg(); TbCallback callback = new TbPackCallback<>(id, ctx); try { ToEdgeMsg toEdgeMsg = msg.getValue(); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java index a809f2bb11..d2a8e7a441 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java @@ -70,6 +70,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractPartitionBasedCo private final TbRuleEngineConsumerContext ctx; private final QueueService queueService; private final TbRuleEngineDeviceRpcService tbDeviceRpcService; + private final TbMsgPackProcessingContextFactory packProcessingContextFactory; private final ConcurrentMap consumers = new ConcurrentHashMap<>(); @@ -85,11 +86,13 @@ public class DefaultTbRuleEngineConsumerService extends AbstractPartitionBasedCo PartitionService partitionService, ApplicationEventPublisher eventPublisher, JwtSettingsService jwtSettingsService, - CalculatedFieldCache calculatedFieldCache) { + CalculatedFieldCache calculatedFieldCache, + TbMsgPackProcessingContextFactory packProcessingContextFactory) { super(actorContext, tenantProfileCache, deviceProfileCache, assetProfileCache, tbResourceDataCache, calculatedFieldCache, apiUsageStateService, partitionService, eventPublisher, jwtSettingsService); this.ctx = ctx; this.tbDeviceRpcService = tbDeviceRpcService; this.queueService = queueService; + this.packProcessingContextFactory = packProcessingContextFactory; } @Override @@ -255,6 +258,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractPartitionBasedCo .consumerExecutor(consumersExecutor) .scheduler(scheduler) .taskExecutor(mgmtExecutor) + .packProcessingContextFactory(packProcessingContextFactory) .build(); consumers.put(queueKey, consumer); consumer.init(queue); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextFactory.java b/application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextFactory.java new file mode 100644 index 0000000000..ff804ad304 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextFactory.java @@ -0,0 +1,35 @@ +/** + * Copyright © 2016-2025 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.queue; + +import org.springframework.stereotype.Component; +import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategy; + +public interface TbMsgPackProcessingContextFactory { + + TbMsgPackProcessingContext create(String queueName, TbRuleEngineSubmitStrategy submitStrategy, boolean skipTimeouts); + + @Component + class DefaultTbMsgPackProcessingContextFactory implements TbMsgPackProcessingContextFactory { + + @Override + public TbMsgPackProcessingContext create(String queueName, TbRuleEngineSubmitStrategy submitStrategy, boolean skipTimeouts) { + return new TbMsgPackProcessingContext(queueName, submitStrategy, skipTimeouts); + } + + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java index e18cc0b09e..7e9ada0e75 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java @@ -134,13 +134,13 @@ public abstract class AbstractConsumerService> msgs, TbQueueConsumer> consumer) throws Exception { List> orderedMsgList = msgs.stream().map(msg -> new IdMsgPair<>(UUID.randomUUID(), msg)).toList(); ConcurrentMap> pendingMap = orderedMsgList.stream().collect( - Collectors.toConcurrentMap(IdMsgPair::getUuid, IdMsgPair::getMsg)); + Collectors.toConcurrentMap(IdMsgPair::uuid, IdMsgPair::msg)); CountDownLatch processingTimeoutLatch = new CountDownLatch(1); TbPackProcessingContext> ctx = new TbPackProcessingContext<>( processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>()); orderedMsgList.forEach(element -> { - UUID id = element.getUuid(); - TbProtoQueueMsg msg = element.getMsg(); + UUID id = element.uuid(); + TbProtoQueueMsg msg = element.msg(); log.trace("[{}] Creating notification callback for message: {}", id, msg.getValue()); TbCallback callback = new TbPackCallback<>(id, ctx); try { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractTbRuleEngineSubmitStrategy.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractTbRuleEngineSubmitStrategy.java index 019573631a..5a1bdae13a 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractTbRuleEngineSubmitStrategy.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractTbRuleEngineSubmitStrategy.java @@ -44,21 +44,21 @@ public abstract class AbstractTbRuleEngineSubmitStrategy implements TbRuleEngine @Override public ConcurrentMap> getPendingMap() { - return orderedMsgList.stream().collect(Collectors.toConcurrentMap(pair -> pair.uuid, pair -> pair.msg)); + return orderedMsgList.stream().collect(Collectors.toConcurrentMap(pair -> pair.uuid(), pair -> pair.msg())); } @Override public void update(ConcurrentMap> reprocessMap) { List> newOrderedMsgList = new ArrayList<>(reprocessMap.size()); for (IdMsgPair pair : orderedMsgList) { - if (reprocessMap.containsKey(pair.uuid)) { - if (StringUtils.isNotEmpty(pair.getMsg().getValue().getFailureMessage())) { - var toRuleEngineMsg = TransportProtos.ToRuleEngineMsg.newBuilder(pair.getMsg().getValue()) + if (reprocessMap.containsKey(pair.uuid())) { + if (StringUtils.isNotEmpty(pair.msg().getValue().getFailureMessage())) { + var toRuleEngineMsg = TransportProtos.ToRuleEngineMsg.newBuilder(pair.msg().getValue()) .clearFailureMessage() .clearRelationTypes() .build(); - var newMsg = new TbProtoQueueMsg<>(pair.getMsg().getKey(), toRuleEngineMsg, pair.getMsg().getHeaders()); - newOrderedMsgList.add(new IdMsgPair<>(pair.getUuid(), newMsg)); + var newMsg = new TbProtoQueueMsg<>(pair.msg().getKey(), toRuleEngineMsg, pair.msg().getHeaders()); + newOrderedMsgList.add(new IdMsgPair<>(pair.uuid(), newMsg)); } else { newOrderedMsgList.add(pair); } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/BatchTbRuleEngineSubmitStrategy.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/BatchTbRuleEngineSubmitStrategy.java index 93478eab18..a47c43422b 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/BatchTbRuleEngineSubmitStrategy.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/BatchTbRuleEngineSubmitStrategy.java @@ -73,7 +73,7 @@ public class BatchTbRuleEngineSubmitStrategy extends AbstractTbRuleEngineSubmitS pendingPack.clear(); for (int i = startIdx; i < endIdx; i++) { IdMsgPair pair = orderedMsgList.get(i); - pendingPack.put(pair.uuid, pair.msg); + pendingPack.put(pair.uuid(), pair.msg()); } tmpPack = new LinkedHashMap<>(pendingPack); } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/BurstTbRuleEngineSubmitStrategy.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/BurstTbRuleEngineSubmitStrategy.java index af465686e9..9f9dd0a60f 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/BurstTbRuleEngineSubmitStrategy.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/BurstTbRuleEngineSubmitStrategy.java @@ -34,7 +34,7 @@ public class BurstTbRuleEngineSubmitStrategy extends AbstractTbRuleEngineSubmitS if (log.isDebugEnabled()) { log.debug("[{}] submitting [{}] messages to rule engine", queueName, orderedMsgList.size()); } - orderedMsgList.forEach(pair -> msgConsumer.accept(pair.uuid, pair.msg)); + orderedMsgList.forEach(pair -> msgConsumer.accept(pair.uuid(), pair.msg())); } @Override diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/IdMsgPair.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/IdMsgPair.java index e74362699b..6795b2822f 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/IdMsgPair.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/IdMsgPair.java @@ -15,19 +15,9 @@ */ package org.thingsboard.server.service.queue.processing; -import lombok.Getter; +import com.google.protobuf.GeneratedMessageV3; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import java.util.UUID; -public class IdMsgPair { - @Getter - final UUID uuid; - @Getter - final TbProtoQueueMsg msg; - - public IdMsgPair(UUID uuid, TbProtoQueueMsg msg) { - this.uuid = uuid; - this.msg = msg; - } -} +public record IdMsgPair(UUID uuid, TbProtoQueueMsg msg) {} diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/SequentialByEntityIdTbRuleEngineSubmitStrategy.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/SequentialByEntityIdTbRuleEngineSubmitStrategy.java index c72f1274e7..1aa1b20284 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/SequentialByEntityIdTbRuleEngineSubmitStrategy.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/SequentialByEntityIdTbRuleEngineSubmitStrategy.java @@ -51,7 +51,7 @@ public abstract class SequentialByEntityIdTbRuleEngineSubmitStrategy extends Abs entityIdToListMap.forEach((entityId, queue) -> { IdMsgPair msg = queue.peek(); if (msg != null) { - msgConsumer.accept(msg.uuid, msg.msg); + msgConsumer.accept(msg.uuid(), msg.msg()); } }); } @@ -71,13 +71,13 @@ public abstract class SequentialByEntityIdTbRuleEngineSubmitStrategy extends Abs IdMsgPair next = null; synchronized (queue) { IdMsgPair expected = queue.peek(); - if (expected != null && expected.uuid.equals(id)) { + if (expected != null && expected.uuid().equals(id)) { queue.poll(); next = queue.peek(); } } if (next != null) { - msgConsumer.accept(next.uuid, next.msg); + msgConsumer.accept(next.uuid(), next.msg()); } } } @@ -87,9 +87,9 @@ public abstract class SequentialByEntityIdTbRuleEngineSubmitStrategy extends Abs msgToEntityIdMap.clear(); entityIdToListMap.clear(); for (IdMsgPair pair : orderedMsgList) { - EntityId entityId = getEntityId(pair.msg.getValue()); + EntityId entityId = getEntityId(pair.msg().getValue()); if (entityId != null) { - msgToEntityIdMap.put(pair.uuid, entityId); + msgToEntityIdMap.put(pair.uuid(), entityId); entityIdToListMap.computeIfAbsent(entityId, id -> new LinkedList<>()).add(pair); } } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/SequentialTbRuleEngineSubmitStrategy.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/SequentialTbRuleEngineSubmitStrategy.java index be97853310..c048f42ee0 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/SequentialTbRuleEngineSubmitStrategy.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/SequentialTbRuleEngineSubmitStrategy.java @@ -60,11 +60,11 @@ public class SequentialTbRuleEngineSubmitStrategy extends AbstractTbRuleEngineSu int idx = msgIdx.get(); if (idx < listSize) { IdMsgPair pair = orderedMsgList.get(idx); - expectedMsgId = pair.uuid; + expectedMsgId = pair.uuid(); if (log.isDebugEnabled()) { - log.debug("[{}] submitting [{}] message to rule engine", queueName, pair.msg); + log.debug("[{}] submitting [{}] message to rule engine", queueName, pair.msg()); } - msgConsumer.accept(pair.uuid, pair.msg); + msgConsumer.accept(pair.uuid(), pair.msg()); } } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java index d067be49a0..a6f7e34208 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java @@ -41,6 +41,7 @@ import org.thingsboard.server.queue.common.consumer.TbQueueConsumerTask; import org.thingsboard.server.queue.discovery.QueueKey; import org.thingsboard.server.service.queue.TbMsgPackCallback; import org.thingsboard.server.service.queue.TbMsgPackProcessingContext; +import org.thingsboard.server.service.queue.TbMsgPackProcessingContextFactory; import org.thingsboard.server.service.queue.TbRuleEngineConsumerStats; import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingDecision; import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingResult; @@ -66,13 +67,15 @@ public class TbRuleEngineQueueConsumerManager extends MainQueueConsumerManager { Integer partitionId = tpi != null ? tpi.getPartition().orElse(-1) : null; @@ -81,6 +84,7 @@ public class TbRuleEngineQueueConsumerManager extends MainQueueConsumerManager submitMessage(packCtx, id, msg)); final boolean timeout = !packCtx.await(queue.getPackProcessingTimeout(), TimeUnit.MILLISECONDS); diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 61b797515a..aa0b5d7920 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1889,6 +1889,9 @@ queue: print-interval-ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:60000}" # Max length of the error message that is printed by statistics max-error-message-length: "${TB_QUEUE_RULE_ENGINE_MAX_ERROR_MESSAGE_LENGTH:4096}" + prometheus-stats: + # Enable/disable Prometheus statistics for individual Rule Engine message processing (records time in ms for success/failure). + enabled: "${TB_QUEUE_RULE_ENGINE_PROMETHEUS_STATS_ENABLED:true}" # After a queue is deleted (or the profile's isolation option was disabled), Rule Engine will continue reading related topics during this period before deleting the actual topics topic-deletion-delay: "${TB_QUEUE_RULE_ENGINE_TOPIC_DELETION_DELAY_SEC:15}" # Size of the thread pool that handles such operations as partition changes, config updates, queue deletion diff --git a/application/src/test/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextTest.java b/application/src/test/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextTest.java index ef1febfa6c..b5d65ea035 100644 --- a/application/src/test/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextTest.java +++ b/application/src/test/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextTest.java @@ -15,14 +15,17 @@ */ package org.thingsboard.server.service.queue; -import lombok.extern.slf4j.Slf4j; -import org.junit.After; -import org.junit.Assert; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.junit.MockitoJUnitRunner; +import com.google.common.util.concurrent.MoreExecutors; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.queue.RuleEngineException; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategy; @@ -35,30 +38,241 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; -import static org.junit.Assert.assertTrue; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.BDDMockito.then; +import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; -@Slf4j -@RunWith(MockitoJUnitRunner.class) -public class TbMsgPackProcessingContextTest { +@ExtendWith(MockitoExtension.class) +class TbMsgPackProcessingContextTest { + + TenantId tenantId = TenantId.fromUUID(UUID.randomUUID()); + + @Mock + TbRuleEngineSubmitStrategy submitStrategy; + @Mock + TbProtoQueueMsg mockMsg; + + ConcurrentMap> pendingMap; - public static final int TIMEOUT = 10; ExecutorService executorService; - @After - public void tearDown() { + @BeforeEach + void setup() { + pendingMap = new ConcurrentHashMap<>(); + lenient().when(submitStrategy.getPendingMap()).thenReturn(pendingMap); + } + + @AfterEach + void tearDown() { if (executorService != null) { - executorService.shutdownNow(); + MoreExecutors.shutdownAndAwaitTermination(executorService, 5, TimeUnit.SECONDS); } } @Test - public void testHighConcurrencyCase() throws InterruptedException { - //log.warn("preparing the test..."); + void testAwait_shouldReturnTrue_whenOnSuccessIsCalledBeforeTimeout() throws InterruptedException { + // GIVEN - a context with one pending message + executorService = Executors.newSingleThreadExecutor(); + + UUID msgId = UUID.randomUUID(); + pendingMap.put(msgId, mockMsg); + var context = new TbMsgPackProcessingContext("test-queue", submitStrategy, false); + + // WHEN - onSuccess() is called in another thread before timeout + executorService.submit(() -> { + try { + Thread.sleep(100); + context.onSuccess(msgId); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + + // THEN - await() should return true (successful completion) + boolean result = context.await(5000, TimeUnit.MILLISECONDS); + assertThat(result).as("await() should return true when latch is counted down before timeout").isTrue(); + + // Verify the message was moved to success map + assertThat(context.getSuccessMap()).containsKey(msgId); + assertThat(context.getPendingMap()).isEmpty(); + assertThat(context.getExceptionsMap()).isEmpty(); + + + // Verify submit strategy was notified about successful message processing + then(submitStrategy).should().onSuccess(msgId); + } + + @Test + void testAwait_shouldReturnTrue_whenOnFailureIsCalledBeforeTimeout() throws InterruptedException { + // GIVEN - a context with one pending message + executorService = Executors.newSingleThreadExecutor(); + + UUID msgId = UUID.randomUUID(); + pendingMap.put(msgId, mockMsg); + var context = new TbMsgPackProcessingContext("test-queue", submitStrategy, false); + + var exception = new RuleEngineException("Test exception"); + + // WHEN - onFailure() is called in another thread before timeout + executorService.submit(() -> { + try { + Thread.sleep(100); + context.onFailure(tenantId, msgId, exception); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + + // THEN - await() should return true (successful completion, even if message processing failed) + boolean result = context.await(5000, TimeUnit.MILLISECONDS); + assertThat(result).as("await() should return true when latch is counted down before timeout").isTrue(); + + // Verify the exception was added to exceptions map + assertThat(context.getSuccessMap()).isEmpty(); + assertThat(context.getPendingMap()).isEmpty(); + assertThat(context.getExceptionsMap()).containsEntry(tenantId, exception); + } + + @Test + void testAwait_shouldReturnFalse_whenTimeoutOccurs() throws InterruptedException { + // GIVEN - a context with one pending message and no processing + UUID msgId = UUID.randomUUID(); + pendingMap.put(msgId, mockMsg); + var context = new TbMsgPackProcessingContext("test-queue", submitStrategy, false); + + // WHEN - await() is called with short timeout and no message processing happens + long startTime = System.nanoTime(); + boolean result = context.await(100, TimeUnit.MILLISECONDS); + long elapsedTime = System.nanoTime() - startTime; + + // THEN - await() should return false (timeout occurred) + assertThat(result).as("await() should return false when timeout occurs").isFalse(); + assertThat(elapsedTime).as("await() should wait for at least the timeout duration").isGreaterThanOrEqualTo(100L); + + // Message should still be in pending map + assertThat(context.getSuccessMap()).isEmpty(); + assertThat(context.getPendingMap()).containsKey(msgId); + assertThat(context.getExceptionsMap()).isEmpty(); + } + + @Test + void testAwait_shouldHandleMultiplePendingMessages() throws InterruptedException { + // GIVEN - a context with multiple pending messages + executorService = Executors.newSingleThreadExecutor(); + + UUID msgId1 = UUID.randomUUID(); + UUID msgId2 = UUID.randomUUID(); + UUID msgId3 = UUID.randomUUID(); + + pendingMap.put(msgId1, mockMsg); + pendingMap.put(msgId2, mockMsg); + pendingMap.put(msgId3, mockMsg); + + var context = new TbMsgPackProcessingContext("test-queue", submitStrategy, false); + + // WHEN - messages are processed one by one + executorService.submit(() -> { + try { + Thread.sleep(50); + context.onSuccess(msgId1); + Thread.sleep(50); + context.onSuccess(msgId2); + Thread.sleep(50); + context.onSuccess(msgId3); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + + // THEN - await() should return true only after all messages are processed + boolean result = context.await(5000, TimeUnit.MILLISECONDS); + assertThat(result).as("await() should return true after all messages are processed").isTrue(); + + // All messages should be in success map + assertThat(context.getSuccessMap()).containsKeys(msgId1, msgId2, msgId3); + assertThat(context.getPendingMap()).isEmpty(); + assertThat(context.getExceptionsMap()).isEmpty(); + } + + @Test + void testAwait_shouldNotCountDownPrematurely_withMultipleMessages() throws InterruptedException { + // GIVEN - a context with multiple pending messages + executorService = Executors.newSingleThreadExecutor(); + + UUID msgId1 = UUID.randomUUID(); + UUID msgId2 = UUID.randomUUID(); + + pendingMap.put(msgId1, mockMsg); + pendingMap.put(msgId2, mockMsg); + + var context = new TbMsgPackProcessingContext("test-queue", submitStrategy, false); + + // WHEN - only one message is processed + executorService.submit(() -> { + try { + Thread.sleep(100); + context.onSuccess(msgId1); + // msgId2 still in processing + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + + // THEN: await should timeout because not all messages were processed + boolean result = context.await(2000, TimeUnit.MILLISECONDS); + assertThat(result).as("await() should timeout when not all messages are processed").isFalse(); + + // One message in success, one still pending + assertThat(context.getSuccessMap()).containsOnlyKeys(msgId1); + assertThat(context.getPendingMap()).containsOnlyKeys(msgId2); + assertThat(context.getExceptionsMap()).isEmpty(); + } + + @Test + void testAwait_shouldHandleMixedSuccessAndFailure() throws InterruptedException { + // GIVEN - multiple messages + executorService = Executors.newSingleThreadExecutor(); + + UUID msgId1 = UUID.randomUUID(); + UUID msgId2 = UUID.randomUUID(); + + pendingMap.put(msgId1, mockMsg); + pendingMap.put(msgId2, mockMsg); + + var context = new TbMsgPackProcessingContext("test-queue", submitStrategy, false); + + var exception = new RuleEngineException("Test exception"); + + // WHEN - one succeeds, one fails + executorService.submit(() -> { + try { + Thread.sleep(50); + context.onSuccess(msgId1); + Thread.sleep(50); + context.onFailure(tenantId, msgId2, exception); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + + // THEN - await() should complete successfully + boolean result = context.await(5000, TimeUnit.MILLISECONDS); + assertThat(result).as("await() should return true when all messages are processed").isTrue(); + + assertThat(context.getSuccessMap()).containsOnlyKeys(msgId1); + assertThat(context.getPendingMap()).isEmpty(); + assertThat(context.getExceptionsMap()).containsEntry(tenantId, exception); + } + + @Test + void testHighConcurrencyCase() throws InterruptedException { int msgCount = 1000; int parallelCount = 5; executorService = Executors.newFixedThreadPool(parallelCount, ThingsBoardThreadFactory.forName(getClass().getSimpleName() + "-test-scope")); @@ -76,28 +290,24 @@ public class TbMsgPackProcessingContextTest { final CountDownLatch startLatch = new CountDownLatch(1); final CountDownLatch finishLatch = new CountDownLatch(parallelCount); for (int i = 0; i < parallelCount; i++) { - //final String taskName = "" + uuid + " " + i; executorService.submit(() -> { - //log.warn("ready {}", taskName); readyLatch.countDown(); try { startLatch.await(); } catch (InterruptedException e) { - Assert.fail("failed to await"); + fail("failed to await"); } - //log.warn("go {}", taskName); - context.onSuccess(uuid); - finishLatch.countDown(); }); } - assertTrue(readyLatch.await(TIMEOUT, TimeUnit.SECONDS)); + assertTrue(readyLatch.await(10, TimeUnit.SECONDS)); Thread.yield(); startLatch.countDown(); //run all-at-once submitted tasks - assertTrue(finishLatch.await(TIMEOUT, TimeUnit.SECONDS)); + assertTrue(finishLatch.await(10, TimeUnit.SECONDS)); } - assertTrue(context.await(TIMEOUT, TimeUnit.SECONDS)); + assertTrue(context.await(10, TimeUnit.SECONDS)); verify(strategyMock, times(msgCount)).onSuccess(any(UUID.class)); } + } diff --git a/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/RuleEngineConsumerLoopTest.java b/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/RuleEngineConsumerLoopTest.java new file mode 100644 index 0000000000..1619283586 --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/RuleEngineConsumerLoopTest.java @@ -0,0 +1,281 @@ +/** + * Copyright © 2016-2025 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.queue.ruleengine; + +import com.google.common.util.concurrent.MoreExecutors; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InOrder; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.thingsboard.common.util.ThingsBoardExecutors; +import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.server.actors.ActorSystemContext; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.msg.TbMsgType; +import org.thingsboard.server.common.data.queue.ProcessingStrategy; +import org.thingsboard.server.common.data.queue.ProcessingStrategyType; +import org.thingsboard.server.common.data.queue.Queue; +import org.thingsboard.server.common.data.queue.SubmitStrategy; +import org.thingsboard.server.common.data.queue.SubmitStrategyType; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.TbMsgMetaData; +import org.thingsboard.server.common.msg.queue.ServiceType; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.common.stats.StatsFactory; +import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.queue.TbQueueAdmin; +import org.thingsboard.server.queue.TbQueueConsumer; +import org.thingsboard.server.queue.TbQueueMsg; +import org.thingsboard.server.queue.common.DefaultTbQueueMsgHeaders; +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.queue.discovery.TbServiceInfoProvider; +import org.thingsboard.server.queue.memory.DefaultInMemoryStorage; +import org.thingsboard.server.queue.memory.InMemoryStorage; +import org.thingsboard.server.queue.memory.InMemoryTbQueueConsumer; +import org.thingsboard.server.queue.provider.TbQueueProducerProvider; +import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory; +import org.thingsboard.server.service.queue.TbMsgPackProcessingContext; +import org.thingsboard.server.service.queue.TbMsgPackProcessingContextFactory; +import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategyFactory; +import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategy; +import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategyFactory; +import org.thingsboard.server.service.stats.RuleEngineStatisticsService; + +import java.time.Duration; +import java.util.List; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +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.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +class RuleEngineConsumerLoopTest { + + TenantId tenantId = TenantId.fromUUID(UUID.randomUUID()); + DeviceId deviceId = new DeviceId(UUID.randomUUID()); + + InMemoryStorage storage; + + @Mock + ActorSystemContext actorContext; + @Mock + StatsFactory statsFactory; + @Mock + TbRuleEngineQueueFactory queueFactory; + @Mock + RuleEngineStatisticsService statisticsService; + @Mock + TbServiceInfoProvider serviceInfoProvider; + @Mock + PartitionService partitionService; + @Mock + TbQueueProducerProvider producerProvider; + @Mock + TbQueueAdmin queueAdmin; + @Mock + TbMsgPackProcessingContextFactory packProcessingContextFactory; + @Mock + TbMsgPackProcessingContext packCtx; + + Queue mainQueue; + + TbQueueConsumer> consumer; + + TbRuleEngineConsumerContext ruleEngineConsumerContext; + TbRuleEngineQueueConsumerManager consumerManager; + + ExecutorService consumersExecutor; + ScheduledExecutorService scheduler; + ExecutorService mgmtExecutor; + + @BeforeEach + void setup() throws InterruptedException { + consumersExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer")); + scheduler = ThingsBoardExecutors.newSingleThreadScheduledExecutor("tb-rule-engine-consumer-scheduler"); + mgmtExecutor = ThingsBoardExecutors.newWorkStealingPool(1, "tb-rule-engine-mgmt"); + + mainQueue = new Queue(); + mainQueue.setTenantId(TenantId.SYS_TENANT_ID); + mainQueue.setName("Main"); + mainQueue.setTopic("tb_rule_engine.main"); + mainQueue.setPollInterval(25); + mainQueue.setPartitions(1); + mainQueue.setConsumerPerPartition(false); + mainQueue.setPackProcessingTimeout(2000L); + + var submitStrategy = new SubmitStrategy(); + submitStrategy.setType(SubmitStrategyType.BURST); + submitStrategy.setBatchSize(1000); + mainQueue.setSubmitStrategy(submitStrategy); + + var processingStrategy = new ProcessingStrategy(); + processingStrategy.setType(ProcessingStrategyType.SKIP_ALL_FAILURES); + processingStrategy.setRetries(3); + processingStrategy.setFailurePercentage(0.0); + processingStrategy.setPauseBetweenRetries(3); + processingStrategy.setMaxPauseBetweenRetries(3); + mainQueue.setProcessingStrategy(processingStrategy); + + storage = new DefaultInMemoryStorage(); + + consumer = spy(new InMemoryTbQueueConsumer<>(storage, mainQueue.getTopic())); + given(queueFactory.createToRuleEngineMsgConsumer(eq(mainQueue), isNull())).willReturn(consumer); + + ruleEngineConsumerContext = new TbRuleEngineConsumerContext( + actorContext, statsFactory, new TbRuleEngineSubmitStrategyFactory(), new TbRuleEngineProcessingStrategyFactory(), + queueFactory, statisticsService, serviceInfoProvider, partitionService, producerProvider, queueAdmin + ); + ruleEngineConsumerContext.setPollDuration(25); + ruleEngineConsumerContext.setPackProcessingTimeout(2000); + ruleEngineConsumerContext.setStatsEnabled(false); // true by default + ruleEngineConsumerContext.setPrometheusStatsEnabled(false); + ruleEngineConsumerContext.setTopicDeletionDelayInSec(15); + ruleEngineConsumerContext.setMgmtThreadPoolSize(12); + + // Tell the (mock) context factory to return (mock) message pack context + given(packProcessingContextFactory.create( + eq(mainQueue.getName()), + any(TbRuleEngineSubmitStrategy.class), + eq(false) + )).willAnswer(invocation -> { + TbRuleEngineSubmitStrategy realStrategy = invocation.getArgument(1); + when(packCtx.getPendingMap()).thenAnswer(i -> realStrategy.getPendingMap()); + when(packCtx.getFailedMap()).thenReturn(new ConcurrentHashMap<>()); + return packCtx; + }); + + // Tell the (mock) context's await() to return 'false' (always timeout) immediately + given(packCtx.await(anyLong(), any(TimeUnit.class))).willReturn(false); + + consumerManager = TbRuleEngineQueueConsumerManager.create() + .ctx(ruleEngineConsumerContext) + .queueKey(new QueueKey(ServiceType.TB_RULE_ENGINE, mainQueue)) + .consumerExecutor(consumersExecutor) + .scheduler(scheduler) + .taskExecutor(mgmtExecutor) + .packProcessingContextFactory(packProcessingContextFactory) + .build(); + } + + @AfterEach + void destroy() { + MoreExecutors.shutdownAndAwaitTermination(scheduler, Duration.ofSeconds(30)); + MoreExecutors.shutdownAndAwaitTermination(mgmtExecutor, Duration.ofSeconds(30)); + MoreExecutors.shutdownAndAwaitTermination(consumersExecutor, Duration.ofSeconds(30)); + } + + @Test + void consumerLoopTest_verifyOperationsOrder() throws InterruptedException { + // Create partition + var partition = TopicPartitionInfo.builder() + .tenantId(TenantId.SYS_TENANT_ID) + .topic(mainQueue.getTopic()) + .partition(0) + .myPartition(true) + .useInternalPartition(false) + .build(); + + // Put 10k messages to the queue + for (int i = 0; i < 10_000; i++) { + var tbMsg = TbMsg.newMsg() + .type(TbMsgType.POST_TELEMETRY_REQUEST) + .originator(deviceId) + .data("{\"temperature\":123}") + .metaData(TbMsgMetaData.EMPTY) + .build(); + + var toRuleEngineMsg = TransportProtos.ToRuleEngineMsg.newBuilder() + .setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) + .setTenantIdMSB(tenantId.getId().getMostSignificantBits()) + .setTbMsgProto(TbMsg.toProto(tbMsg)) + .addAllRelationTypes(Set.of("Success")) + .build(); + + storage.put(partition.getFullTopicName(), new TbProtoQueueMsg<>(UUID.randomUUID(), toRuleEngineMsg, new DefaultTbQueueMsgHeaders())); + } + + // Count how many polls were made + var totalPolls = new AtomicInteger(0); + var emptyPolls = new AtomicInteger(0); + doAnswer(invocation -> { + totalPolls.incrementAndGet(); + @SuppressWarnings("unchecked") + var messages = (List) invocation.callRealMethod(); + if (messages.isEmpty()) { + emptyPolls.incrementAndGet(); + } + return messages; + }).when(consumer).poll(mainQueue.getPollInterval()); + + // Count how many commits were made + var totalCommits = new AtomicInteger(0); + doAnswer(invocation -> { + totalCommits.incrementAndGet(); + return invocation.callRealMethod(); + }).when(consumer).commit(); + + // Initialize consumer + consumerManager.init(mainQueue); + + // Assign partition to the consumer + consumerManager.update(Set.of(partition)); + + // Give some time for the consumer to get all messages + await().atMost(Duration.ofSeconds(10L)).until(() -> storage.getLagTotal() == 0); + + // Stop consumer + consumerManager.stop(); + consumerManager.awaitStop(); + + // Determine number of non-empty consumer iterations made, since polling does not stop immediately after consuming all messages and may do a few empty polls + int nonEmptyPolls = totalPolls.get() - emptyPolls.get(); + + // Verify that there is 10 polls and 10 matching commits + // Each poll consumes 1k messages and queue has 10k total, so that means 10k total msgs / 1k msgs per poll = 10 polls + assertThat(nonEmptyPolls).isEqualTo(10).isEqualTo(totalCommits.get()); + + // Verify that poll-await-commit cycle happened in order with correct await timeout + InOrder inOrder = inOrder(consumer, packCtx); + for (int i = 0; i < nonEmptyPolls; i++) { + inOrder.verify(consumer).poll(mainQueue.getPollInterval()); + inOrder.verify(packCtx).await(mainQueue.getPackProcessingTimeout(), TimeUnit.MILLISECONDS); + inOrder.verify(consumer).commit(); + } + } + +} From 56a83b6cfec7794a92cae6889722d0bb5e565d0d Mon Sep 17 00:00:00 2001 From: Dmytro Skarzhynets Date: Tue, 28 Oct 2025 16:19:05 +0200 Subject: [PATCH 2/3] Fix existing test --- .../queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java | 2 ++ .../service/queue/ruleengine/TbRuleEngineStrategyTest.java | 2 ++ 2 files changed, 4 insertions(+) diff --git a/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java b/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java index bcbe52b5c9..3c54d421e4 100644 --- a/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java +++ b/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java @@ -59,6 +59,7 @@ import org.thingsboard.server.queue.provider.KafkaMonolithQueueFactory; import org.thingsboard.server.queue.provider.KafkaTbRuleEngineQueueFactory; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory; +import org.thingsboard.server.service.queue.TbMsgPackProcessingContextFactory; import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategyFactory; import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategyFactory; import org.thingsboard.server.service.stats.RuleEngineStatisticsService; @@ -194,6 +195,7 @@ public class TbRuleEngineQueueConsumerManagerTest { .consumerExecutor(consumersExecutor) .scheduler(scheduler) .taskExecutor(mgmtExecutor) + .packProcessingContextFactory(new TbMsgPackProcessingContextFactory.DefaultTbMsgPackProcessingContextFactory()) .build(); } diff --git a/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineStrategyTest.java b/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineStrategyTest.java index 1106fad5b6..c27a03619a 100644 --- a/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineStrategyTest.java +++ b/application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineStrategyTest.java @@ -44,6 +44,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.QueueKey; +import org.thingsboard.server.service.queue.TbMsgPackProcessingContextFactory; import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategyFactory; import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategyFactory; @@ -194,6 +195,7 @@ public class TbRuleEngineStrategyTest { var consumerManager = TbRuleEngineQueueConsumerManager.create() .ctx(ruleEngineConsumerContext) .queueKey(queueKey) + .packProcessingContextFactory(new TbMsgPackProcessingContextFactory.DefaultTbMsgPackProcessingContextFactory()) .build(); consumerManager.init(queue); From cb2bd954b6a99e32596efc9be84d16040d18fab8 Mon Sep 17 00:00:00 2001 From: Dmytro Skarzhynets Date: Wed, 3 Dec 2025 13:20:15 +0200 Subject: [PATCH 3/3] Fix default value for Prometheus rule engine stats --- application/src/main/resources/thingsboard.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 2ee0d23502..2cb53a8a4b 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1913,7 +1913,7 @@ queue: max-error-message-length: "${TB_QUEUE_RULE_ENGINE_MAX_ERROR_MESSAGE_LENGTH:4096}" prometheus-stats: # Enable/disable Prometheus statistics for individual Rule Engine message processing (records time in ms for success/failure). - enabled: "${TB_QUEUE_RULE_ENGINE_PROMETHEUS_STATS_ENABLED:true}" + enabled: "${TB_QUEUE_RULE_ENGINE_PROMETHEUS_STATS_ENABLED:false}" # After a queue is deleted (or the profile's isolation option was disabled), Rule Engine will continue reading related topics during this period before deleting the actual topics topic-deletion-delay: "${TB_QUEUE_RULE_ENGINE_TOPIC_DELETION_DELAY_SEC:15}" # Size of the thread pool that handles such operations as partition changes, config updates, queue deletion