added upgrade for queues

application/src/main/data/upgrade/3.3.4/schema_update.sql

@@ -0,0 +1,30 @@
+--
+-- Copyright © 2016-2022 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.
+--
+
+CREATE TABLE IF NOT EXISTS queue (
+    id uuid NOT NULL CONSTRAINT queue_pkey PRIMARY KEY,
+    created_time bigint NOT NULL,
+    tenant_id uuid,
+    name varchar(255),
+    topic varchar(255),
+    poll_interval int,
+    partitions int,
+    consumer_per_partition boolean,
+    pack_processing_timeout bigint,
+    submit_strategy varchar(255),
+    processing_strategy varchar(255),
+    additional_info varchar
+);

application/src/main/data/upgrade/3.3.4/schema_update_device_profile.sql

@@ -0,0 +1,49 @@
+--
+-- Copyright © 2016-2022 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.
+--
+
+ALTER TABLE device_profile
+    ADD COLUMN IF NOT EXISTS default_queue_id uuid;
+
+DO
+$$
+    BEGIN
+        IF NOT EXISTS(SELECT 1 FROM pg_constraint WHERE conname = 'fk_default_queue_device_profile') THEN
+            ALTER TABLE device_profile
+                ADD CONSTRAINT fk_default_queue_device_profile FOREIGN KEY (default_queue_id) REFERENCES queue (id);
+        END IF;
+    END;
+$$;
+
+DO
+$$
+    BEGIN
+        IF EXISTS
+            (SELECT column_name
+             FROM information_schema.columns
+             WHERE table_name = 'device_profile'
+               AND column_name = 'default_queue_name'
+            )
+        THEN
+            UPDATE device_profile
+            SET default_queue_id = q.id
+            FROM queue as q
+            WHERE default_queue_name = q.name;
+        END IF;
+    END
+$$;
+
+ALTER TABLE device_profile
+    DROP COLUMN IF EXISTS default_queue_name;

application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java

@@ -219,6 +219,7 @@ public class ThingsboardInstallService {
                             databaseEntitiesUpgradeService.upgradeDatabase("3.3.3");
                         case "3.3.4":
                             log.info("Upgrading ThingsBoard from version 3.3.4 to 3.4.0 ...");
+                            databaseEntitiesUpgradeService.upgradeDatabase("3.3.4");
                             log.info("Updating system data...");
                             systemDataLoaderService.updateSystemWidgets();
                             break;

application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java

@@ -15,18 +15,31 @@
  */
 package org.thingsboard.server.service.install;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Profile;
 import org.springframework.stereotype.Service;
 import org.thingsboard.server.common.data.EntitySubtype;
 import org.thingsboard.server.common.data.Tenant;
+import org.thingsboard.server.common.data.id.QueueId;
+import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.page.PageLink;
+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.data.rule.RuleNode;
 import org.thingsboard.server.dao.dashboard.DashboardService;
 import org.thingsboard.server.dao.device.DeviceProfileService;
 import org.thingsboard.server.dao.device.DeviceService;
+import org.thingsboard.server.dao.queue.QueueService;
+import org.thingsboard.server.dao.rule.RuleChainService;
 import org.thingsboard.server.dao.tenant.TenantService;
 import org.thingsboard.server.dao.usagerecord.ApiUsageStateService;
 import org.thingsboard.server.service.install.sql.SqlDbHelper;
@@ -43,7 +56,9 @@ import java.sql.SQLSyntaxErrorException;
 import java.sql.SQLWarning;
 import java.sql.Statement;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 import static org.thingsboard.server.service.install.DatabaseHelper.ADDITIONAL_INFO;
 import static org.thingsboard.server.service.install.DatabaseHelper.ASSIGNED_CUSTOMERS;
@@ -101,6 +116,14 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService
     @Autowired
     private ApiUsageStateService apiUsageStateService;
 
+    @Autowired
+    private QueueService queueService;
+
+    @Autowired
+    private TbRuleEngineQueueConfigService queueConfig;
+
+    @Autowired
+    private RuleChainService ruleChainService;
 
     @Override
     public void upgradeDatabase(String fromVersion) throws Exception {
@@ -534,6 +557,81 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService
                     log.error("Failed updating schema!!!", e);
                 }
                 break;
+            case "3.3.4":
+                try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) {
+                    log.info("Updating schema ...");
+                    schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "3.3.4", SCHEMA_UPDATE_SQL);
+                    loadSql(schemaUpdateFile, conn);
+
+                    log.info("Loading queues...");
+                    try {
+                        if (!CollectionUtils.isEmpty(queueConfig.getQueues())) {
+                            queueConfig.getQueues().forEach(queueSettings -> {
+                                Queue queue = new Queue();
+                                queue.setTenantId(TenantId.SYS_TENANT_ID);
+                                queue.setName(queueSettings.getName());
+                                queue.setTopic(queueSettings.getTopic());
+                                queue.setPollInterval(queueSettings.getPollInterval());
+                                queue.setPartitions(queueSettings.getPartitions());
+                                queue.setPackProcessingTimeout(queueSettings.getPackProcessingTimeout());
+                                SubmitStrategy submitStrategy = new SubmitStrategy();
+                                submitStrategy.setBatchSize(queueSettings.getSubmitStrategy().getBatchSize());
+                                submitStrategy.setType(SubmitStrategyType.valueOf(queueSettings.getSubmitStrategy().getType()));
+                                queue.setSubmitStrategy(submitStrategy);
+                                ProcessingStrategy processingStrategy = new ProcessingStrategy();
+                                processingStrategy.setType(ProcessingStrategyType.valueOf(queueSettings.getProcessingStrategy().getType()));
+                                processingStrategy.setRetries(queueSettings.getProcessingStrategy().getRetries());
+                                processingStrategy.setFailurePercentage(queueSettings.getProcessingStrategy().getFailurePercentage());
+                                processingStrategy.setPauseBetweenRetries(queueSettings.getProcessingStrategy().getPauseBetweenRetries());
+                                processingStrategy.setMaxPauseBetweenRetries(queueSettings.getProcessingStrategy().getMaxPauseBetweenRetries());
+                                queue.setProcessingStrategy(processingStrategy);
+                                queue.setConsumerPerPartition(queueSettings.isConsumerPerPartition());
+                                queueService.saveQueue(queue);
+                            });
+                        } else {
+                            systemDataLoaderService.createQueues();
+                        }
+                    } catch (Exception e) {
+                    }
+
+                    log.info("Updating device profiles...");
+                    schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "3.3.4", "schema_update_device_profile.sql");
+                    loadSql(schemaUpdateFile, conn);
+
+                    log.info("Updating checkpoint rule nodes...");
+                    PageLink pageLink = new PageLink(100);
+                    PageData<Tenant> pageData;
+                    do {
+                        pageData = tenantService.findTenants(pageLink);
+                        for (Tenant tenant : pageData.getData()) {
+                            TenantId tenantId = tenant.getId();
+                            Map<String, QueueId> queues =
+                                    queueService.findQueuesByTenantId(tenantId).stream().collect(Collectors.toMap(Queue::getName, Queue::getId));
+                            try {
+                                List<RuleNode> checkpointNodes =
+                                        ruleChainService.findRuleNodesByTenantIdAndType(tenantId, "org.thingsboard.rule.engine.flow.TbCheckpointNode");
+                                checkpointNodes.forEach(node -> {
+                                    ObjectNode configuration = (ObjectNode) node.getConfiguration();
+                                    JsonNode queueNameNode = configuration.remove("queueName");
+                                    if (queueNameNode != null) {
+                                        String queueName = queueNameNode.asText();
+                                        configuration.put("queueId", queues.get(queueName).toString());
+                                        ruleChainService.saveRuleNode(tenantId, node);
+                                    }
+                                });
+                            } catch (Exception e) {
+                            }
+                        }
+                        pageLink = pageLink.nextPageLink();
+                    } while (pageData.hasNext());
+
+                    log.info("Updating schema settings...");
+                    conn.createStatement().execute("UPDATE tb_schema_settings SET schema_version = 3004000;");
+                    log.info("Schema updated.");
+                } catch (Exception e) {
+                    log.error("Failed updating schema!!!", e);
+                }
+                break;
             default:
                 throw new RuntimeException("Unable to upgrade SQL database, unsupported fromVersion: " + fromVersion);
         }

application/src/main/java/org/thingsboard/server/service/install/TbRuleEngineQueueConfigService.java

@@ -0,0 +1,48 @@
+/**
+ * Copyright © 2016-2022 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.install;
+
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
+import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
+
+import javax.annotation.PostConstruct;
+import java.util.List;
+
+@Slf4j
+@Data
+@EnableAutoConfiguration
+@Configuration
+@ConfigurationProperties(prefix = "queue.rule-engine")
+@Profile("install")
+public class TbRuleEngineQueueConfigService {
+
+    private String topic;
+    private List<TbRuleEngineQueueConfiguration> queues;
+
+    @PostConstruct
+    public void validate() {
+        queues.stream().filter(queue -> queue.getName().equals("Main")).findFirst().orElseThrow(() -> {
+            log.error("Main queue is not configured in thingsboard.yml");
+            return new RuntimeException("No \"Main\" queue configured!");
+        });
+    }
+
+}

application/src/main/java/org/thingsboard/server/service/install/update/DefaultCacheCleanupService.java

@@ -68,6 +68,10 @@ public class DefaultCacheCleanupService implements CacheCleanupService {
                 log.info("Clear cache to upgrade from version 3.3.3 to 3.3.4 ...");
                 clearAll();
                 break;
+            case "3.3.4":
+                log.info("Clear cache to upgrade from version 3.3.4 to 3.4.0 ...");
+                clearCacheByName("deviceProfiles");
+                break;
             default:
                 //Do nothing, since cache cleanup is optional.
         }

common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueServiceDeprecated.java

@@ -1,28 +0,0 @@
-/**
- * Copyright © 2016-2022 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;
-
-import org.thingsboard.server.common.msg.queue.ServiceType;
-
-import java.util.Set;
-
-public interface TbQueueServiceDeprecated {
-
-    Set<String> getQueuesByServiceType(ServiceType serviceType);
-
-    String resolve(ServiceType serviceType, String queueName);
-
-}

common/dao-api/src/main/java/org/thingsboard/server/dao/rule/RuleChainService.java

@@ -92,5 +92,7 @@ public interface RuleChainService {
 
     List<RuleNode> findRuleNodesByTenantIdAndType(TenantId tenantId, String name, String toString);
 
+    List<RuleNode> findRuleNodesByTenantIdAndType(TenantId tenantId, String type);
+
     RuleNode saveRuleNode(TenantId tenantId, RuleNode ruleNode);
 }

common/queue/src/main/java/org/thingsboard/server/queue/DefaultTbQueueServiceDeprecated.java

@@ -1,61 +0,0 @@
-/**
- * Copyright © 2016-2022 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;
-
-import lombok.RequiredArgsConstructor;
-import org.springframework.util.StringUtils;
-import org.thingsboard.server.common.msg.queue.ServiceQueue;
-import org.thingsboard.server.common.msg.queue.ServiceType;
-import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
-import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
-
-import javax.annotation.PostConstruct;
-import java.util.Collections;
-import java.util.LinkedHashSet;
-import java.util.Set;
-import java.util.stream.Collectors;
-
-//@Service
-@RequiredArgsConstructor
-public class DefaultTbQueueServiceDeprecated implements TbQueueServiceDeprecated {
-
-    private final TbQueueRuleEngineSettings ruleEngineSettings;
-    private Set<String> ruleEngineQueues;
-
-    @PostConstruct
-    public void init() {
-        ruleEngineQueues = ruleEngineSettings.getQueues().stream()
-                .map(TbRuleEngineQueueConfiguration::getName).collect(Collectors.toCollection(LinkedHashSet::new));
-    }
-
-    @Override
-    public Set<String> getQueuesByServiceType(ServiceType type) {
-        if (type == ServiceType.TB_RULE_ENGINE) {
-            return ruleEngineQueues;
-        } else {
-            return Collections.emptySet();
-        }
-    }
-
-    @Override
-    public String resolve(ServiceType serviceType, String queueName) {
-        if (StringUtils.isEmpty(queueName) || !getQueuesByServiceType(serviceType).contains(queueName)) {
-            return ServiceQueue.MAIN;
-        } else {
-            return queueName;
-        }
-    }
-}

common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueRuleEngineSettings.java

@@ -16,32 +16,14 @@
 package org.thingsboard.server.queue.settings;
 
 import lombok.Data;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
-import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.context.annotation.Configuration;
+import org.springframework.stereotype.Component;
 
-import javax.annotation.PostConstruct;
-import java.util.List;
-
-@Slf4j
 @Data
-@EnableAutoConfiguration
-@Configuration
-@ConfigurationProperties(prefix = "queue.rule-engine")
+@Component
 public class TbQueueRuleEngineSettings {
 
+    @Value("${queue.rule-engine.topic}")
     private String topic;
-    private List<TbRuleEngineQueueConfiguration> queues;
-
-    @PostConstruct
-    public void validate() {
-        queues.stream().filter(queue -> queue.getName().equals("Main")).findFirst().orElseThrow(() -> {
-            log.error("Main queue is not configured in thingsboard.yml");
-            return new RuntimeException("No \"Main\" queue configured!");
-        });
-    }
 
 }

dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java

@@ -668,6 +668,14 @@ public class BaseRuleChainService extends AbstractEntityService implements RuleC
         return ruleNodeDao.findRuleNodesByTenantIdAndType(tenantId, type, search);
     }
 
+    @Override
+    public List<RuleNode> findRuleNodesByTenantIdAndType(TenantId tenantId, String type) {
+        log.trace("Executing findRuleNodes, tenantId [{}], type {}", tenantId, type);
+        validateId(tenantId, INCORRECT_TENANT_ID + tenantId);
+        validateString(type, "Incorrect type of the rule node");
+        return ruleNodeDao.findRuleNodesByTenantIdAndType(tenantId, type, "");
+    }
+
     @Override
     public RuleNode saveRuleNode(TenantId tenantId, RuleNode ruleNode) {
         return ruleNodeDao.save(tenantId, ruleNode);

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbCheckpointNode.java

@@ -24,7 +24,6 @@ import org.thingsboard.rule.engine.api.TbNodeException;
 import org.thingsboard.rule.engine.api.TbRelationTypes;
 import org.thingsboard.rule.engine.api.util.TbNodeUtils;
 import org.thingsboard.server.common.data.plugin.ComponentType;
-import org.thingsboard.server.common.data.queue.Queue;
 import org.thingsboard.server.common.msg.TbMsg;
 
 @Slf4j
@@ -44,12 +43,6 @@ public class TbCheckpointNode implements TbNode {
     @Override
     public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
         this.config = TbNodeUtils.convert(configuration, TbCheckpointNodeConfiguration.class);
-        if (config.getQueueId() == null) {
-            Queue foundQueue = ctx.getQueueService().findQueueByTenantIdAndName(ctx.getTenantId(), config.getQueueName());
-            if (foundQueue != null) {
-                config.setQueueId(foundQueue.getId());
-            }
-        }
     }
 
     @Override

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbCheckpointNodeConfiguration.java

@@ -22,14 +22,10 @@ import org.thingsboard.server.common.data.id.QueueId;
 @Data
 public class TbCheckpointNodeConfiguration implements NodeConfiguration<TbCheckpointNodeConfiguration> {
 
-    private String queueName;
-
     private QueueId queueId;
 
     @Override
     public TbCheckpointNodeConfiguration defaultConfiguration() {
-        TbCheckpointNodeConfiguration configuration = new TbCheckpointNodeConfiguration();
-        configuration.setQueueName("HighPriority");
-        return configuration;
+        return new TbCheckpointNodeConfiguration();
     }
 }