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 a2bed09151..e1bd18c7bf 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 @@ -127,8 +127,8 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService, CoreQueueConfig>builder() .queueKey(new QueueKey(ServiceType.TB_CORE)) - .config(CoreQueueConfig.of(consumerPerPartitionEnabled, (int) pollInterval)) + .config(CoreQueueConfig.of(consumerPerPartition, (int) pollInterval)) .msgPackProcessor(this::processMsgs) .consumerCreator(config -> queueFactory.createToCoreMsgConsumer()) .consumerExecutor(consumersExecutor) @@ -325,7 +325,9 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService log.debug("[{}] Timeout to process message: {}", id, msg.getValue())); + if (log.isDebugEnabled()) { + ctx.getAckMap().forEach((id, msg) -> log.debug("[{}] Timeout to process message: {}", id, msg.getValue())); + } ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process message: {}", id, msg.getValue())); } consumer.commit(); @@ -452,6 +454,8 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService { } catch (Throwable e) { log.error("Failure in consumer loop", e); } + log.info("[{}] Consumer stopped", name); }); } @@ -94,7 +95,6 @@ public class BasicQueueConsumerManager { } } } - log.info("{} Consumer stopped", name); } public void stop() { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java b/application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java index 04538289ea..acd28f25ba 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java @@ -176,8 +176,9 @@ public class QueueConsumerManager { doUpdate(partitions); // even if partitions number was changed, there can be no partition change event } } else { + log.trace("[{}] Silently applied new config, because consumer-per-partition not changed", queueKey); // do nothing, because partitions change (if they changed) will be handled on PartitionChangeEvent, - // and changes to pollInterval/packProcessingTimeout/submitStrategy/processingStrategy will be picked up by consumer on the fly, + // and changes to other config values will be picked up by consumer on the fly, // and queue topic and name are immutable } } @@ -196,6 +197,7 @@ public class QueueConsumerManager { } catch (Throwable e) { log.error("Failure in consumer loop", e); } + log.info("[{}] Consumer stopped", consumerTask.getKey()); }); consumerTask.setTask(consumerLoop); } @@ -222,7 +224,6 @@ public class QueueConsumerManager { if (consumer.isStopped()) { consumer.unsubscribe(); } - log.info("{} Consumer stopped", queueKey); } protected void processMsgs(List msgs, TbQueueConsumer consumer, C config) throws Exception { @@ -236,6 +237,7 @@ public class QueueConsumerManager { } public void awaitStop() { + log.debug("[{}] Waiting for consumers to stop", queueKey); consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::awaitCompletion); log.debug("[{}] Unsubscribed and stopped consumers", queueKey); } @@ -272,7 +274,7 @@ public class QueueConsumerManager { removedPartitions.forEach((tpi) -> consumers.remove(tpi).awaitCompletion()); addedPartitions.forEach((tpi) -> { - String key = queueKey + "-" + tpi.getPartition().orElse(-999999); + String key = queueKey + "-" + tpi.getPartition().orElse(-1); TbQueueConsumerTask consumer = new TbQueueConsumerTask<>(key, consumerCreator.apply(config)); consumers.put(tpi, consumer); consumer.subscribe(Set.of(tpi)); 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 fa43eaa78d..79942cf6df 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 @@ -15,6 +15,7 @@ */ package org.thingsboard.server.service.queue.processing; +import jakarta.annotation.PreDestroy; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.context.ApplicationEventPublisher; @@ -48,7 +49,6 @@ import org.thingsboard.server.service.queue.TbPackProcessingContext; import org.thingsboard.server.service.queue.consumer.BasicQueueConsumerManager; import org.thingsboard.server.service.security.auth.jwt.settings.JwtSettingsService; -import jakarta.annotation.PreDestroy; import java.util.List; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index d38c83a9ef..ca186ed0a5 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1596,7 +1596,7 @@ queue: # Timeout for processing a message pack by Core microservices pack-processing-timeout: "${TB_QUEUE_CORE_PACK_PROCESSING_TIMEOUT_MS:2000}" # Enable/disable a separate consumer per partition for Core queue - consumer-per-partition-enabled: "${TB_QUEUE_CORE_CONSUMER_PER_PARTITION_ENABLED:true}" + consumer-per-partition: "${TB_QUEUE_CORE_CONSUMER_PER_PARTITION:true}" ota: # Default topic name for OTA updates topic: "${TB_QUEUE_CORE_OTA_TOPIC:tb_ota_package}" 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 a00f45b8db..75df04333d 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 @@ -453,7 +453,7 @@ public class TbRuleEngineQueueConsumerManagerTest { int msgCount = totalConsumedMsgs.get(); await().atLeast(2, TimeUnit.SECONDS) // based on topicDeletionDelayInSec(5) = 5 - ( 3 seconds the code may execute starting consumerManager.delete() call) - .atMost(10, TimeUnit.SECONDS) + .atMost(20, TimeUnit.SECONDS) .untilAsserted(() -> { partitions.stream() .map(TopicPartitionInfo::getFullTopicName) @@ -492,7 +492,7 @@ public class TbRuleEngineQueueConsumerManagerTest { int msgCount = totalConsumedMsgs.get(); await().atLeast(2, TimeUnit.SECONDS) // based on topicDeletionDelayInSec(5) = 5 - ( 3 seconds the code may execute starting consumerManager.delete() call) - .atMost(10, TimeUnit.SECONDS) + .atMost(20, TimeUnit.SECONDS) .untilAsserted(() -> { partitions.stream() .map(TopicPartitionInfo::getFullTopicName)