Merge branch 'master' into feature/bar-chart-with-labels

3 years ago · 8c398620f8
18 changed files with 91 additions and 39 deletions
--- a/application/src/main/data/upgrade/3.6.2/schema_update.sql
+++ b/application/src/main/data/upgrade/3.6.2/schema_update.sql
@ -0,0 +1,23 @@
+--
+-- Copyright © 2016-2023 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.
+--
+
+-- RULE NODE INDEXES UPDATE START
+
+DROP INDEX IF EXISTS idx_rule_node_type;
+DROP INDEX IF EXISTS idx_rule_node_type_configuration_version;
+CREATE INDEX IF NOT EXISTS idx_rule_node_type_id_configuration_version ON rule_node(type, id, configuration_version);
+
+-- RULE NODE INDEXES UPDATE END
--- a/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java
+++ b/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java
@ -276,6 +276,9 @@ public class ThingsboardInstallService {
                            } else {
                                log.info("Skipping images migration. Run the upgrade with fromVersion as '3.6.2-images' to migrate");
                            }
+                        case "3.6.2":
+                            log.info("Upgrading ThingsBoard from version 3.6.2 to 3.6.3 ...");
+                            databaseEntitiesUpgradeService.upgradeDatabase("3.6.2");
                            //TODO DON'T FORGET to update switch statement in the CacheCleanupService if you need to clear the cache
                            break;
                        default:
--- a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java
+++ b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java
@ -772,6 +772,9 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService
                    }
                });
                break;
+            case "3.6.2":
+                updateSchema("3.6.2", 3006002, "3.6.3", 3006003, null);
+                break;
            default:
                throw new RuntimeException("Unable to upgrade SQL database, unsupported fromVersion: " + fromVersion);
        }
--- a/application/src/main/resources/thingsboard.yml
+++ b/application/src/main/resources/thingsboard.yml
@ -1398,6 +1398,8 @@ queue:
    region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}"
    # Number of threads per each AWS SQS queue in consumer
    threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}"
+    # Thread pool size for aws_sqs queue producer executor provider. Default value equals to AmazonSQSAsyncClient.DEFAULT_THREAD_POOL_SIZE
+    producer_thread_pool_size: "${TB_QUEUE_AWS_SQS_EXECUTOR_THREAD_POOL_SIZE:50}"
    queue-properties:
      # AWS SQS queue properties. VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds
      rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}"
--- a/common/dao-api/src/main/java/org/thingsboard/server/dao/rule/RuleChainService.java
+++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/rule/RuleChainService.java
@ -100,6 +100,7 @@ public interface RuleChainService extends EntityDaoService {

    PageData<RuleNode> findAllRuleNodesByType(String type, PageLink pageLink);

+    @Deprecated(forRemoval = true, since = "3.6.3")
    PageData<RuleNode> findAllRuleNodesByTypeAndVersionLessThan(String type, int version, PageLink pageLink);

    PageData<RuleNodeId> findAllRuleNodeIdsByTypeAndVersionLessThan(String type, int version, PageLink pageLink);
--- a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsAdmin.java
+++ b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsAdmin.java
@ -24,11 +24,17 @@ import com.amazonaws.services.sqs.AmazonSQS;
 import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
 import com.amazonaws.services.sqs.model.CreateQueueRequest;
 import com.amazonaws.services.sqs.model.GetQueueUrlResult;
+import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.common.util.ExecutorProvider;
+import org.thingsboard.common.util.ThingsBoardExecutors;
+import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.server.queue.TbQueueAdmin;
 import org.thingsboard.server.queue.util.PropertyUtils;

 import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.function.Function;
 import java.util.stream.Collectors;

@ -38,6 +44,8 @@ public class TbAwsSqsAdmin implements TbQueueAdmin {
    private final Map<String, String> attributes;
    private final AmazonSQS sqsClient;
    private final Map<String, String> queues;
+    @Getter
+    private final ExecutorService producerExecutor;

    public TbAwsSqsAdmin(TbAwsSqsSettings sqsSettings, Map<String, String> attributes) {
        this.attributes = attributes;
@ -49,6 +57,7 @@ public class TbAwsSqsAdmin implements TbQueueAdmin {
            AWSCredentials awsCredentials = new BasicAWSCredentials(sqsSettings.getAccessKeyId(), sqsSettings.getSecretAccessKey());
            credentialsProvider = new AWSStaticCredentialsProvider(awsCredentials);
        }
+        producerExecutor = ThingsBoardExecutors.newWorkStealingPool(sqsSettings.getThreadPoolSize(), "aws-sqs-queue-executor");

        sqsClient = AmazonSQSClientBuilder.standard()
                .withCredentials(credentialsProvider)
@ -104,5 +113,8 @@ public class TbAwsSqsAdmin implements TbQueueAdmin {
        if (sqsClient != null) {
            sqsClient.shutdown();
        }
+        if (producerExecutor != null) {
+            producerExecutor.shutdownNow();
+        }
    }
 }
--- a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsProducerTemplate.java
+++ b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsProducerTemplate.java
@ -20,15 +20,11 @@ import com.amazonaws.auth.AWSCredentialsProvider;
 import com.amazonaws.auth.AWSStaticCredentialsProvider;
 import com.amazonaws.auth.BasicAWSCredentials;
 import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
-import com.amazonaws.services.sqs.AmazonSQS;
-import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
+import com.amazonaws.handlers.AsyncHandler;
+import com.amazonaws.services.sqs.AmazonSQSAsync;
+import com.amazonaws.services.sqs.AmazonSQSAsyncClientBuilder;
 import com.amazonaws.services.sqs.model.SendMessageRequest;
 import com.amazonaws.services.sqs.model.SendMessageResult;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.ListeningExecutorService;
-import com.google.common.util.concurrent.MoreExecutors;
 import com.google.gson.Gson;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
@ -41,19 +37,17 @@ import org.thingsboard.server.queue.common.DefaultTbQueueMsg;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;

@Slf4j
 public class TbAwsSqsProducerTemplate<T extends TbQueueMsg> implements TbQueueProducer<T> {
    private final String defaultTopic;
-    private final AmazonSQS sqsClient;
+    private final AmazonSQSAsync sqsClient;
    private final Gson gson = new Gson();
    private final Map<String, String> queueUrlMap = new ConcurrentHashMap<>();
-    private final TbQueueAdmin admin;
-    private ListeningExecutorService producerExecutor;
+    private final TbAwsSqsAdmin admin;

    public TbAwsSqsProducerTemplate(TbQueueAdmin admin, TbAwsSqsSettings sqsSettings, String defaultTopic) {
-        this.admin = admin;
+        this.admin = (TbAwsSqsAdmin) admin;
        this.defaultTopic = defaultTopic;

        AWSCredentialsProvider credentialsProvider;
@ -64,11 +58,11 @@ public class TbAwsSqsProducerTemplate<T extends TbQueueMsg> implements TbQueuePr
            credentialsProvider = new AWSStaticCredentialsProvider(awsCredentials);
        }

-        sqsClient = AmazonSQSClientBuilder.standard()
+        sqsClient = AmazonSQSAsyncClientBuilder.standard()
                .withCredentials(credentialsProvider)
                .withRegion(sqsSettings.getRegion())
+                .withExecutorFactory(this.admin::getProducerExecutor)
                .build();
-        producerExecutor = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());
    }

    @Override
@ -91,30 +85,24 @@ public class TbAwsSqsProducerTemplate<T extends TbQueueMsg> implements TbQueuePr
        sendMsgRequest.withMessageGroupId(sqsMsgId);
        sendMsgRequest.withMessageDeduplicationId(sqsMsgId);

-        ListenableFuture<SendMessageResult> future = producerExecutor.submit(() -> sqsClient.sendMessage(sendMsgRequest));
-
-        Futures.addCallback(future, new FutureCallback<SendMessageResult>() {
-            @Override
-            public void onSuccess(SendMessageResult result) {
+        sqsClient.sendMessageAsync(sendMsgRequest, new AsyncHandler<SendMessageRequest, SendMessageResult>() {
+            @Override public void onError(Exception e) {
                if (callback != null) {
-                    callback.onSuccess(new AwsSqsTbQueueMsgMetadata(result.getSdkHttpMetadata()));
+                    callback.onFailure(e);
                }
            }

-            @Override
-            public void onFailure(Throwable t) {
+            @Override public void onSuccess(SendMessageRequest request,
+                                            SendMessageResult sendMessageResult) {
                if (callback != null) {
-                    callback.onFailure(t);
+                    callback.onSuccess(new AwsSqsTbQueueMsgMetadata(sendMessageResult.getSdkHttpMetadata()));
                }
            }
-        }, producerExecutor);
+        });
    }

    @Override
    public void stop() {
-        if (producerExecutor != null) {
-            producerExecutor.shutdownNow();
-        }
        if (sqsClient != null) {
            sqsClient.shutdown();
        }
--- a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsSettings.java
+++ b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsSettings.java
@ -42,4 +42,7 @@ public class TbAwsSqsSettings {
    @Value("${queue.aws_sqs.threads_per_topic}")
    private int threadsPerTopic;

+    @Value("${queue.aws_sqs.producer_thread_pool_size:50}")
+    private int threadPoolSize;
+
 }
--- a/common/script/script-api/src/test/java/org/thingsboard/script/api/tbel/TbDateTest.java
+++ b/common/script/script-api/src/test/java/org/thingsboard/script/api/tbel/TbDateTest.java
@ -177,8 +177,9 @@ class TbDateTest {
         * For Java 17:
         * `{ "AM", "PM" }`
         */
-        s = "09:15:30 nachm., So. 10/09/2022";
-        d = new TbDate(s, pattern, "de","Europe/Berlin");
+        String s_ver = Runtime.version().feature() == 11 ? "09:15:30 nachm., So. 10/09/2022" :
+                                                           "09:15:30 PM, So. 10/09/2022";
+        d = new TbDate(s_ver, pattern, "de","Europe/Berlin");
        Assert.assertEquals("2022-10-09T19:15:30Z", d.toISOString());

        s = "02:15:30 пп, середа, 4 жовтня 2023 р.";
@ -295,7 +296,9 @@ class TbDateTest {
        Assert.assertEquals("9/5/23, 9:04:05 PM", d.toLocaleString("en-US", "America/New_York"));
        Assert.assertEquals("23. 9. 5. 오후 9:04:05", d.toLocaleString("ko-KR",  "America/New_York"));
        Assert.assertEquals("06.09.23, 04:04:05",  d.toLocaleString( "uk-UA", "Europe/Kiev"));
-        Assert.assertEquals("5\u200F/9\u200F/2023 9:04:05 م",  d.toLocaleString( "ar-EG", "America/New_York"));
+        String expected_ver = Runtime.version().feature() == 11 ? "5\u200F/9\u200F/2023 9:04:05 م" :
+                                                                  "5\u200F/9\u200F/2023, 9:04:05 م";
+        Assert.assertEquals(expected_ver,  d.toLocaleString( "ar-EG", "America/New_York"));

        Assert.assertEquals("Tuesday, September 5, 2023 at 9:04:05 PM Eastern Daylight Time", d.toLocaleString("en-US", JacksonUtil.newObjectNode()
                .put("timeZone", "America/New_York")
@ -311,8 +314,11 @@ class TbDateTest {
                .put("timeZone", "Europe/Kiev")
                .put("dateStyle", "full")
                .put("timeStyle", "full")
-                .toString())); 
-        Assert.assertEquals("الثلاثاء، 5 سبتمبر 2023 9:04:05 م التوقيت الصيفي الشرقي لأمريكا الشمالية", d.toLocaleString("ar-EG", JacksonUtil.newObjectNode()
+                .toString()));
+
+        expected_ver = Runtime.version().feature() == 11 ? "الثلاثاء، 5 سبتمبر 2023 9:04:05 م التوقيت الصيفي الشرقي لأمريكا الشمالية" :
+                                                           "الثلاثاء، 5 سبتمبر 2023 في 9:04:05 م التوقيت الصيفي الشرقي لأمريكا الشمالية";
+        Assert.assertEquals(expected_ver, d.toLocaleString("ar-EG", JacksonUtil.newObjectNode()
                .put("timeZone", "America/New_York")
                .put("dateStyle", "full")
                .put("timeStyle", "full")
@ -381,8 +387,9 @@ class TbDateTest {

        // With pattern + locale - ok
        String pattern = "hh:mm:ss a, EEE M/d/uuuu";
-        stringDateRFC_1123 = "09:15:30 nachm., So. 10/09/2022";
-        d = new TbDate(stringDateRFC_1123 , pattern, "de");
+        String stringDate_ver_RFC_1123 = Runtime.version().feature() == 11 ? "09:15:30 nachm., So. 10/09/2022" :
+                                                                             "09:15:30 PM, So. 10/09/2022";
+        d = new TbDate(stringDate_ver_RFC_1123 , pattern, "de");
        Assert.assertEquals("2022-10-09 21:15:30", d.toLocaleString());

            // failed TZ
--- a/dao/src/main/java/org/thingsboard/server/dao/service/Validator.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/service/Validator.java
@ -61,7 +61,7 @@ public class Validator {


    /**
-     * This method validate <code>long</code> value. If value isn't possitive than throw
+     * This method validate <code>long</code> value. If value isn't positive than throw
     * <code>IncorrectParameterException</code> exception
     *
     * @param val          the val
--- a/dao/src/main/java/org/thingsboard/server/dao/sql/rule/RuleNodeRepository.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rule/RuleNodeRepository.java
@ -43,7 +43,7 @@ public interface RuleNodeRepository extends JpaRepository<RuleNodeEntity, UUID>
                                                Pageable pageable);

    @Query(nativeQuery = true, value = "SELECT * FROM rule_node r WHERE r.type = :ruleType " +
-            " AND configuration_version < :version " +
+            " AND r.configuration_version < :version " +
            " AND (:searchText IS NULL OR r.configuration ILIKE CONCAT('%', :searchText, '%'))")
    Page<RuleNodeEntity> findAllRuleNodesByTypeAndVersionLessThan(@Param("ruleType") String ruleType,
                                                                  @Param("version") int version,
--- a/dao/src/main/resources/sql/schema-entities-idx.sql
+++ b/dao/src/main/resources/sql/schema-entities-idx.sql
@ -91,9 +91,7 @@ CREATE INDEX IF NOT EXISTS idx_widgets_bundle_external_id ON widgets_bundle(tena

 CREATE INDEX IF NOT EXISTS idx_rule_node_external_id ON rule_node(rule_chain_id, external_id);

-CREATE INDEX IF NOT EXISTS idx_rule_node_type ON rule_node(type);
-
-CREATE INDEX IF NOT EXISTS idx_rule_node_type_configuration_version ON rule_node(type, configuration_version);
+CREATE INDEX IF NOT EXISTS idx_rule_node_type_id_configuration_version ON rule_node(type, id, configuration_version);

 CREATE INDEX IF NOT EXISTS idx_api_usage_state_entity_id ON api_usage_state(entity_id);

--- a/msa/vc-executor/src/main/resources/tb-vc-executor.yml
+++ b/msa/vc-executor/src/main/resources/tb-vc-executor.yml
@ -155,6 +155,8 @@ queue:
    region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}"
    # Number of threads per each AWS SQS queue in consumer
    threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}"
+    # Thread pool size for aws_sqs queue producer executor provider. Default value equals to AmazonSQSAsyncClient.DEFAULT_THREAD_POOL_SIZE
+    producer_thread_pool_size: "${TB_QUEUE_AWS_SQS_EXECUTOR_THREAD_POOL_SIZE:50}"
    queue-properties:
      # AWS SQS queue properties. VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds
      core: "${TB_QUEUE_AWS_SQS_CORE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}"
--- a/transport/coap/src/main/resources/tb-coap-transport.yml
+++ b/transport/coap/src/main/resources/tb-coap-transport.yml
@ -281,6 +281,8 @@ queue:
    region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}"
    # Number of threads per each AWS SQS queue in consumer
    threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}"
+    # Thread pool size for aws_sqs queue producer executor provider. Default value equals to AmazonSQSAsyncClient.DEFAULT_THREAD_POOL_SIZE
+    producer_thread_pool_size: "${TB_QUEUE_AWS_SQS_EXECUTOR_THREAD_POOL_SIZE:50}"
    queue-properties:
      # AWS SQS queue properties. VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds
      rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}"
--- a/transport/http/src/main/resources/tb-http-transport.yml
+++ b/transport/http/src/main/resources/tb-http-transport.yml
@ -265,6 +265,8 @@ queue:
    region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}"
    # Number of threads per each AWS SQS queue in consumer
    threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}"
+    # Thread pool size for aws_sqs queue producer executor provider. Default value equals to AmazonSQSAsyncClient.DEFAULT_THREAD_POOL_SIZE
+    producer_thread_pool_size: "${TB_QUEUE_AWS_SQS_EXECUTOR_THREAD_POOL_SIZE:50}"
    queue-properties:
      # AWS SQS queue properties. VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds
      rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}"
--- a/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml
+++ b/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml
@ -360,6 +360,8 @@ queue:
    region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}"
    # Number of threads per each AWS SQS queue in consumer
    threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}"
+    # Thread pool size for aws_sqs queue producer executor provider. Default value equals to AmazonSQSAsyncClient.DEFAULT_THREAD_POOL_SIZE
+    producer_thread_pool_size: "${TB_QUEUE_AWS_SQS_EXECUTOR_THREAD_POOL_SIZE:50}"
    queue-properties:
      # AWS SQS queue properties. VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds
      rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}"
--- a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml
+++ b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml
@ -297,6 +297,8 @@ queue:
    region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}"
    # Number of threads per each AWS SQS queue in consumer
    threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}"
+    # Thread pool size for aws_sqs queue producer executor provider. Default value equals to AmazonSQSAsyncClient.DEFAULT_THREAD_POOL_SIZE
+    producer_thread_pool_size: "${TB_QUEUE_AWS_SQS_EXECUTOR_THREAD_POOL_SIZE:50}"
    queue-properties:
      # AWS SQS queue properties. VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds
      rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}"
--- a/transport/snmp/src/main/resources/tb-snmp-transport.yml
+++ b/transport/snmp/src/main/resources/tb-snmp-transport.yml
@ -250,6 +250,8 @@ queue:
    region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}"
    # Number of threads per each AWS SQS queue in consumer
    threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}"
+    # Thread pool size for aws_sqs queue producer executor provider. Default value equals to AmazonSQSAsyncClient.DEFAULT_THREAD_POOL_SIZE
+    producer_thread_pool_size: "${TB_QUEUE_AWS_SQS_EXECUTOR_THREAD_POOL_SIZE:50}"
    queue-properties:
      # AWS SQS queue properties. VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds
      rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}"