AWS improvements

common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsConsumerTemplate.java

@@ -127,6 +127,11 @@ public class TbAwsSqsConsumerTemplate<T extends TbQueueMsg> implements TbQueueCo
             if (!subscribed) {
                 List<String> topicNames = partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.toList());
                 queueUrls = topicNames.stream().map(this::getQueueUrl).collect(Collectors.toSet());
+
+                if (consumerExecutor != null) {
+                    consumerExecutor.shutdown();
+                }
+
                 consumerExecutor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(queueUrls.size() * sqsSettings.getThreadsPerTopic() + 1));
                 subscribed = true;
             }
@@ -172,7 +177,6 @@ public class TbAwsSqsConsumerTemplate<T extends TbQueueMsg> implements TbQueueCo
                 ReceiveMessageRequest request = new ReceiveMessageRequest();
                 request
                         .withWaitTimeSeconds(waitTimeSeconds)
-                        .withMessageAttributeNames("headers")
                         .withQueueUrl(url)
                         .withMaxNumberOfMessages(MAX_NUM_MSGS);
                 return sqsClient.receiveMessage(request).getMessages();

common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsProducerTemplate.java

@@ -37,6 +37,7 @@ import org.thingsboard.server.queue.TbQueueProducer;
 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;
 
@@ -80,7 +81,9 @@ public class TbAwsSqsProducerTemplate<T extends TbQueueMsg> implements TbQueuePr
         sendMsgRequest.withQueueUrl(getQueueUrl(tpi.getFullTopicName()));
         sendMsgRequest.withMessageBody(gson.toJson(new DefaultTbQueueMsg(msg)));
 
-        sendMsgRequest.withMessageGroupId(msg.getKey().toString());
+        sendMsgRequest.withMessageGroupId(tpi.getTopic());
+        sendMsgRequest.withMessageDeduplicationId(UUID.randomUUID().toString());
+
         ListenableFuture<SendMessageResult> future = producerExecutor.submit(() -> sqsClient.sendMessage(sendMsgRequest));
 
         Futures.addCallback(future, new FutureCallback<SendMessageResult>() {

common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsQueueAttributes.java

@@ -55,7 +55,6 @@ public class TbAwsSqsQueueAttributes {
     @PostConstruct
     private void init() {
         defaultAttributes.put(QueueAttributeName.FifoQueue.toString(), "true");
-        defaultAttributes.put(QueueAttributeName.ContentBasedDeduplication.toString(), "true");
 
         coreAttributes = getConfigs(coreProperties);
         ruleEngineAttributes = getConfigs(ruleEngineProperties);

msa/js-executor/package.json

@@ -22,6 +22,7 @@
     "azure-sb": "^0.11.1",
     "long": "^4.0.0",
     "uuid-parse": "^1.0.0",
+    "uuid-random": "^1.3.0",
     "winston": "^3.0.0",
     "winston-daily-rotate-file": "^3.2.1"
   },

msa/js-executor/queue/awsSqsTemplate.js

@@ -19,6 +19,7 @@
 const config = require('config'),
     JsInvokeMessageProcessor = require('../api/jsInvokeMessageProcessor'),
     logger = require('../config/logger')._logger('awsSqsTemplate');
+const uuid = require('uuid-random');
 
 const requestTopic = config.get('request_topic');
 
@@ -29,7 +30,7 @@ const AWS = require('aws-sdk');
 const queueProperties = config.get('aws_sqs.queue_properties');
 const poolInterval = config.get('js.response_poll_interval');
 
-let queueAttributes = {FifoQueue: 'true', ContentBasedDeduplication: 'true'};
+let queueAttributes = {FifoQueue: 'true'};
 let sqsClient;
 let requestQueueURL;
 const queueUrls = new Map();
@@ -51,7 +52,7 @@ function AwsSqsProducer() {
             queueUrls.set(responseTopic, responseQueueUrl);
         }
 
-        let params = {MessageBody: msgBody, QueueUrl: responseQueueUrl, MessageGroupId: scriptId};
+        let params = {MessageBody: msgBody, QueueUrl: responseQueueUrl, MessageGroupId: 'js_eval', MessageDeduplicationId: uuid()};
 
         return new Promise((resolve, reject) => {
             sqsClient.sendMessage(params, function (err, data) {

msa/js-executor/queue/pubSubTemplate.js

@@ -60,9 +60,11 @@ function PubSubProducer() {
         const topicList = await pubSubClient.getTopics();
 
         if (topicList) {
-            topicList[0].forEach(topic => {
-                topics.push(getName(topic.name));
-            });
+            if (topicList) {
+                topicList[0].forEach(topic => {
+                    topics.push(getName(topic.name));
+                });
+            }
         }
 
         const subscriptionList = await pubSubClient.getSubscriptions();