From 0f1ef6d0dc3ba641afc56099d48f7469e86b2888 Mon Sep 17 00:00:00 2001 From: Sergey Matvienko Date: Wed, 22 May 2024 16:09:32 +0200 Subject: [PATCH] consumer creation refactored after merge --- .../queue/DefaultTbCoreConsumerService.java | 2 +- .../consumer/MainQueueConsumerManager.java | 13 +-- .../TbRuleEngineQueueConsumerManager.java | 83 ------------------- .../TbRuleEngineQueueConsumerManagerTest.java | 4 +- 4 files changed, 10 insertions(+), 92 deletions(-) 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 fbef0cfd29..ae8ea8f7f4 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 @@ -204,7 +204,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService queueFactory.createToCoreMsgConsumer()) + .consumerCreator((config, partitionId) -> queueFactory.createToCoreMsgConsumer()) .consumerExecutor(consumersExecutor) .scheduler(scheduler) .taskExecutor(mgmtExecutor) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/consumer/MainQueueConsumerManager.java b/application/src/main/java/org/thingsboard/server/service/queue/consumer/MainQueueConsumerManager.java index 6792422b57..6eb5c94c9b 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/consumer/MainQueueConsumerManager.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/consumer/MainQueueConsumerManager.java @@ -41,7 +41,7 @@ import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.ReentrantLock; -import java.util.function.Function; +import java.util.function.BiFunction; import java.util.stream.Collectors; @Slf4j @@ -51,7 +51,7 @@ public class MainQueueConsumerManager msgPackProcessor; - protected final Function> consumerCreator; + protected final BiFunction> consumerCreator; protected final ExecutorService consumerExecutor; protected final ScheduledExecutorService scheduler; protected final ExecutorService taskExecutor; @@ -67,7 +67,7 @@ public class MainQueueConsumerManager msgPackProcessor, - Function> consumerCreator, + BiFunction> consumerCreator, ExecutorService consumerExecutor, ScheduledExecutorService scheduler, ExecutorService taskExecutor) { @@ -273,8 +273,9 @@ public class MainQueueConsumerManager consumers.remove(tpi).awaitCompletion()); addedPartitions.forEach((tpi) -> { - String key = queueKey + "-" + tpi.getPartition().orElse(-1); - TbQueueConsumerTask consumer = new TbQueueConsumerTask<>(key, consumerCreator.apply(config)); + Integer partitionId = tpi.getPartition().orElse(-1); + String key = queueKey + "-" + partitionId; + TbQueueConsumerTask consumer = new TbQueueConsumerTask<>(key, () -> consumerCreator.apply(config, partitionId)); consumers.put(tpi, consumer); consumer.subscribe(Set.of(tpi)); launchConsumer(consumer); @@ -303,7 +304,7 @@ public class MainQueueConsumerManager(queueKey, consumerCreator.apply(config)); + consumer = new TbQueueConsumerTask<>(queueKey, () -> consumerCreator.apply(config, null)); // no partitionId passed } consumer.subscribe(partitions); if (!consumer.isRunning()) { 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 8a96f2c0e1..c2823d3c00 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 @@ -249,87 +249,4 @@ public class TbRuleEngineQueueConsumerManager extends MainQueueConsumerManager partitions) { - return partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.joining(", ", "[", "]")); - } - - interface ConsumerWrapper { - - void updatePartitions(Set partitions); - - Collection getConsumers(); - - } - - class ConsumerPerPartitionWrapper implements ConsumerWrapper { - private final Map consumers = new HashMap<>(); - - @Override - public void updatePartitions(Set partitions) { - Set addedPartitions = new HashSet<>(partitions); - addedPartitions.removeAll(consumers.keySet()); - - Set removedPartitions = new HashSet<>(consumers.keySet()); - removedPartitions.removeAll(partitions); - log.info("[{}] Added partitions: {}, removed partitions: {}", queueKey, partitionsToString(addedPartitions), partitionsToString(removedPartitions)); - - removedPartitions.forEach((tpi) -> { - consumers.get(tpi).initiateStop(); - }); - - List removedTasks = removedPartitions.stream() - .map(consumers::remove).toList(); - -// ctx.getQueueAdmin().createTopicIfNotExists(); - addedPartitions.forEach((tpi) -> { - int partitionId = tpi.getPartition().orElse(-999999); - String key = queueKey + "-" + partitionId; - TbQueueConsumerTask consumer = new TbQueueConsumerTask(key, () -> ctx.getQueueFactory().createToRuleEngineMsgConsumer(queue, partitionId)); - consumers.put(tpi, consumer); - consumer.subscribe(Set.of(tpi)); - launchConsumer(consumer); - }); - - removedTasks.forEach(TbQueueConsumerTask::awaitCompletion); - } - - @Override - public Collection getConsumers() { - return consumers.values(); - } - } - - class SingleConsumerWrapper implements ConsumerWrapper { - private TbQueueConsumerTask consumer; - - @Override - public void updatePartitions(Set partitions) { - log.info("[{}] New partitions: {}", queueKey, partitionsToString(partitions)); - if (partitions.isEmpty()) { - if (consumer != null && consumer.isRunning()) { - consumer.initiateStop(); - consumer.awaitCompletion(); - } - consumer = null; - return; - } - - if (consumer == null) { - consumer = new TbQueueConsumerTask(queueKey, () -> ctx.getQueueFactory().createToRuleEngineMsgConsumer(queue, null)); - } - consumer.subscribe(partitions); - if (!consumer.isRunning()) { - launchConsumer(consumer); - } - } - - @Override - public Collection getConsumers() { - if (consumer == null) { - return Collections.emptyList(); - } - return List.of(consumer); - } - } - } 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 35b5eaeb51..fdf2e203ee 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 @@ -21,10 +21,10 @@ import lombok.extern.slf4j.Slf4j; 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.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import org.mockito.Mock; import org.mockito.Mockito; -import org.mockito.junit.jupiter.MockitoExtension; import org.mockito.junit.jupiter.MockitoSettings; import org.mockito.quality.Strictness; import org.testcontainers.shaded.org.apache.commons.lang3.RandomUtils;