Merge branch 'rc' into feature/add_show_total_legend_setting_to_latest-chart-widgets

1 year ago · b008ee5c46
8 changed files with 19 additions and 51 deletions
--- a/application/src/main/java/org/thingsboard/server/service/edqs/KafkaEdqsSyncService.java
+++ b/application/src/main/java/org/thingsboard/server/service/edqs/KafkaEdqsSyncService.java
@ -18,6 +18,7 @@ package org.thingsboard.server.service.edqs;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.stereotype.Service;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
+import org.thingsboard.server.queue.discovery.TopicService;
 import org.thingsboard.server.queue.edqs.EdqsConfig;
 import org.thingsboard.server.queue.kafka.TbKafkaAdmin;
 import org.thingsboard.server.queue.kafka.TbKafkaSettings;
@ -32,11 +33,11 @@ public class KafkaEdqsSyncService extends EdqsSyncService {

    private final boolean syncNeeded;

-    public KafkaEdqsSyncService(TbKafkaSettings kafkaSettings, EdqsConfig edqsConfig) {
+    public KafkaEdqsSyncService(TbKafkaSettings kafkaSettings, TopicService topicService, EdqsConfig edqsConfig) {
        TbKafkaAdmin kafkaAdmin = new TbKafkaAdmin(kafkaSettings, Collections.emptyMap());
        this.syncNeeded = kafkaAdmin.areAllTopicsEmpty(IntStream.range(0, edqsConfig.getPartitions())
                .mapToObj(partition -> TopicPartitionInfo.builder()
-                        .topic(edqsConfig.getEventsTopic())
+                        .topic(topicService.buildTopicName(edqsConfig.getEventsTopic()))
                        .partition(partition)
                        .build().getFullTopicName())
                .collect(Collectors.toSet()));
--- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
+++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
@ -251,7 +251,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
        mainConsumer.update(event.getCorePartitions());
        usageStatsConsumer.subscribe(event.getCorePartitions()
                .stream()
-                .map(tpi -> tpi.newByTopic(usageStatsConsumer.getConsumer().getTopic()))
+                .map(tpi -> tpi.withTopic(usageStatsConsumer.getConsumer().getTopic()))
                .collect(Collectors.toSet()));
    }

--- a/common/edqs/src/main/java/org/thingsboard/server/edqs/processor/EdqsProcessor.java
+++ b/common/edqs/src/main/java/org/thingsboard/server/edqs/processor/EdqsProcessor.java
@ -58,6 +58,7 @@ import org.thingsboard.server.queue.TbQueueResponseTemplate;
 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.common.consumer.PartitionedQueueConsumerManager;
 import org.thingsboard.server.queue.discovery.QueueKey;
+import org.thingsboard.server.queue.discovery.TopicService;
 import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
 import org.thingsboard.server.queue.edqs.EdqsComponent;
 import org.thingsboard.server.queue.edqs.EdqsConfig;
@ -88,6 +89,7 @@ public class EdqsProcessor implements TbQueueHandler<TbProtoQueueMsg<ToEdqsMsg>,
    private final EdqsRepository repository;
    private final EdqsConfig config;
    private final EdqsPartitionService partitionService;
+    private final TopicService topicService;
    private final ConfigurableApplicationContext applicationContext;
    private final EdqsStateService stateService;

@ -123,7 +125,7 @@ public class EdqsProcessor implements TbQueueHandler<TbProtoQueueMsg<ToEdqsMsg>,

        eventConsumer = PartitionedQueueConsumerManager.<TbProtoQueueMsg<ToEdqsMsg>>create()
                .queueKey(new QueueKey(ServiceType.EDQS, config.getEventsTopic()))
-                .topic(config.getEventsTopic())
+                .topic(topicService.buildTopicName(config.getEventsTopic()))
                .pollInterval(config.getPollInterval())
                .msgPackProcessor((msgs, consumer, config) -> {
                    for (TbProtoQueueMsg<ToEdqsMsg> queueMsg : msgs) {
@ -164,9 +166,9 @@ public class EdqsProcessor implements TbQueueHandler<TbProtoQueueMsg<ToEdqsMsg>,
        try {
            Set<TopicPartitionInfo> newPartitions = event.getNewPartitions().get(new QueueKey(ServiceType.EDQS));

-            stateService.process(withTopic(newPartitions, config.getStateTopic()));
+            stateService.process(withTopic(newPartitions, topicService.buildTopicName(config.getStateTopic())));
            // eventsConsumer's partitions are updated by stateService
-            responseTemplate.subscribe(withTopic(newPartitions, config.getRequestsTopic())); // TODO: we subscribe to partitions before we are ready. implement consumer-per-partition version for request template
+            responseTemplate.subscribe(withTopic(newPartitions, topicService.buildTopicName(config.getRequestsTopic()))); // TODO: we subscribe to partitions before we are ready. implement consumer-per-partition version for request template

            Set<TopicPartitionInfo> oldPartitions = event.getOldPartitions().get(new QueueKey(ServiceType.EDQS));
            if (CollectionsUtil.isNotEmpty(oldPartitions)) {
--- a/common/edqs/src/main/java/org/thingsboard/server/edqs/state/KafkaEdqsStateService.java
+++ b/common/edqs/src/main/java/org/thingsboard/server/edqs/state/KafkaEdqsStateService.java
@ -36,6 +36,7 @@ import org.thingsboard.server.queue.common.consumer.QueueConsumerManager;
 import org.thingsboard.server.queue.common.state.KafkaQueueStateService;
 import org.thingsboard.server.queue.common.state.QueueStateService;
 import org.thingsboard.server.queue.discovery.QueueKey;
+import org.thingsboard.server.queue.discovery.TopicService;
 import org.thingsboard.server.queue.edqs.EdqsConfig;
 import org.thingsboard.server.queue.edqs.KafkaEdqsComponent;
 import org.thingsboard.server.queue.edqs.KafkaEdqsQueueFactory;
@ -59,6 +60,7 @@ public class KafkaEdqsStateService implements EdqsStateService {
    private final EdqsConfig config;
    private final EdqsPartitionService partitionService;
    private final KafkaEdqsQueueFactory queueFactory;
+    private final TopicService topicService;
    @Autowired
    @Lazy
    private EdqsProcessor edqsProcessor;
@ -78,7 +80,7 @@ public class KafkaEdqsStateService implements EdqsStateService {
        TbKafkaAdmin queueAdmin = queueFactory.getEdqsQueueAdmin();
        stateConsumer = PartitionedQueueConsumerManager.<TbProtoQueueMsg<ToEdqsMsg>>create()
                .queueKey(new QueueKey(ServiceType.EDQS, config.getStateTopic()))
-                .topic(config.getStateTopic())
+                .topic(topicService.buildTopicName(config.getStateTopic()))
                .pollInterval(config.getPollInterval())
                .msgPackProcessor((msgs, consumer, config) -> {
                    for (TbProtoQueueMsg<ToEdqsMsg> queueMsg : msgs) {
@ -176,7 +178,7 @@ public class KafkaEdqsStateService implements EdqsStateService {
        if (queueStateService.getPartitions().isEmpty()) {
            Set<TopicPartitionInfo> allPartitions = IntStream.range(0, config.getPartitions())
                    .mapToObj(partition -> TopicPartitionInfo.builder()
-                            .topic(config.getEventsTopic())
+                            .topic(topicService.buildTopicName(config.getEventsTopic()))
                            .partition(partition)
                            .build())
                    .collect(Collectors.toSet());
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/TopicPartitionInfo.java
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/TopicPartitionInfo.java
@ -57,10 +57,6 @@ public class TopicPartitionInfo {
        this(topic, tenantId, partition, false, myPartition);
    }

-    public TopicPartitionInfo newByTopic(String topic) {
-        return new TopicPartitionInfo(topic, this.tenantId, this.partition, this.useInternalPartition, this.myPartition);
-    }
-
    public String getTopic() {
        return topic;
    }
--- 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
@ -156,7 +156,7 @@ public class HashPartitionService implements PartitionService {

    @Override
    public String getTopic(QueueKey queueKey) {
-        return partitionTopicsMap.get(queueKey);
+        return topicService.buildTopicName(partitionTopicsMap.get(queueKey));
    }

    private void doInitRuleEnginePartitions() {
--- a/common/queue/src/main/java/org/thingsboard/server/queue/usagestats/DefaultTbApiUsageReportClient.java
+++ b/common/queue/src/main/java/org/thingsboard/server/queue/usagestats/DefaultTbApiUsageReportClient.java
@ -131,7 +131,7 @@ public class DefaultTbApiUsageReportClient implements TbApiUsageReportClient {
        report.forEach((parent, statsMsg) -> {
            try {
                TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, parent.getTenantId(), parent.getId())
-                        .newByTopic(msgProducer.getDefaultTopic());
+                        .withTopic(msgProducer.getDefaultTopic());
                reportStatsPerTpi.computeIfAbsent(tpi, k -> new ArrayList<>()).add(statsMsg.build());
            } catch (TenantNotFoundException e) {
                log.debug("Couldn't report usage stats for non-existing tenant: {}", e.getTenantId());
--- a/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java
+++ b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java
@ -29,7 +29,6 @@ import java.nio.file.Path;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@ -46,7 +45,6 @@ public class ContainerTestSuite {
    final static boolean IS_REDIS_SENTINEL = Boolean.parseBoolean(System.getProperty("blackBoxTests.redisSentinel"));
    final static boolean IS_REDIS_SSL = Boolean.parseBoolean(System.getProperty("blackBoxTests.redisSsl"));
    final static boolean IS_HYBRID_MODE = Boolean.parseBoolean(System.getProperty("blackBoxTests.hybridMode"));
-    final static String QUEUE_TYPE = System.getProperty("blackBoxTests.queue", "kafka");
    private static final String SOURCE_DIR = "./../../docker/";
    private static final String TB_CORE_LOG_REGEXP = ".*Starting polling for events.*";
    private static final String TRANSPORTS_LOG_REGEXP = ".*Going to recalculate partitions.*";
@ -122,44 +120,13 @@ public class ContainerTestSuite {
                    new File(targetDir + (IS_HYBRID_MODE ? "docker-compose.hybrid.yml" : "docker-compose.postgres.yml")),
                    new File(targetDir + (IS_HYBRID_MODE ? "docker-compose.hybrid-test-extras.yml" : "docker-compose.postgres-test-extras.yml")),
                    new File(targetDir + "docker-compose.postgres.volumes.yml"),
-                    new File(targetDir + "docker-compose." + QUEUE_TYPE + ".yml"),
+                    new File(targetDir + "docker-compose.kafka.yml"),
                    new File(targetDir + resolveRedisComposeFile()),
                    new File(targetDir + resolveRedisComposeVolumesFile()),
                    new File(targetDir + ("docker-selenium.yml"))
            ));
-
-            Map<String, String> queueEnv = new HashMap<>();
-            queueEnv.put("TB_QUEUE_TYPE", QUEUE_TYPE);
-            switch (QUEUE_TYPE) {
-                case "kafka":
-                    composeFiles.add(new File(targetDir + "docker-compose.kafka.yml"));
-                    break;
-                case "aws-sqs":
-                    replaceInFile(targetDir, "queue-aws-sqs.env",
-                            Map.of("YOUR_KEY", getSysProp("blackBoxTests.awsKey"),
-                                    "YOUR_SECRET", getSysProp("blackBoxTests.awsSecret"),
-                                    "YOUR_REGION", getSysProp("blackBoxTests.awsRegion")));
-                    break;
-                case "rabbitmq":
-                    composeFiles.add(new File(targetDir + "docker-compose.rabbitmq-server.yml"));
-                    replaceInFile(targetDir, "queue-rabbitmq.env",
-                            Map.of("localhost", "rabbitmq"));
-                    break;
-                case "service-bus":
-                    replaceInFile(targetDir, "queue-service-bus.env",
-                            Map.of("YOUR_NAMESPACE_NAME", getSysProp("blackBoxTests.serviceBusNamespace"),
-                                    "YOUR_SAS_KEY_NAME", getSysProp("blackBoxTests.serviceBusSASPolicy")));
-                    replaceInFile(targetDir, "queue-service-bus.env",
-                            Map.of("YOUR_SAS_KEY", getSysProp("blackBoxTests.serviceBusPrimaryKey")));
-                    break;
-                case "pubsub":
-                    replaceInFile(targetDir, "queue-pubsub.env",
-                            Map.of("YOUR_PROJECT_ID", getSysProp("blackBoxTests.pubSubProjectId"),
-                                    "YOUR_SERVICE_ACCOUNT", getSysProp("blackBoxTests.pubSubServiceAccount")));
-                    break;
-                default:
-                    throw new RuntimeException("Unsupported queue type: " + QUEUE_TYPE);
-            }
+            addToFile(targetDir, "queue-kafka.env", Map.of("TB_QUEUE_PREFIX", "test"));
+            addToFile(targetDir, "tb-edqs.env", Map.of("TB_QUEUE_PREFIX", "test"));

            if (IS_HYBRID_MODE) {
                composeFiles.add(new File(targetDir + "docker-compose.cassandra.volumes.yml"));
@ -171,7 +138,7 @@ public class ContainerTestSuite {
                    .withOptions("--compatibility")
                    .withTailChildContainers(!skipTailChildContainers)
                    .withEnv(installTb.getEnv())
-                    .withEnv(queueEnv)
+                    .withEnv("TB_QUEUE_TYPE", "kafka")
                    .withEnv("LOAD_BALANCER_NAME", "")
                    .withExposedService("haproxy", 80, Wait.forHttp("/swagger-ui.html").withStartupTimeout(CONTAINER_STARTUP_TIMEOUT))
                    .withExposedService("broker", 1883)