refactored

common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubAdmin.java

@@ -15,6 +15,7 @@
  */
 package org.thingsboard.server.queue.pubsub;
 
+import com.google.api.gax.rpc.AlreadyExistsException;
 import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
 import com.google.cloud.pubsub.v1.SubscriptionAdminSettings;
 import com.google.cloud.pubsub.v1.TopicAdminClient;
@@ -24,9 +25,9 @@ import com.google.pubsub.v1.ListSubscriptionsRequest;
 import com.google.pubsub.v1.ListTopicsRequest;
 import com.google.pubsub.v1.ProjectName;
 import com.google.pubsub.v1.ProjectSubscriptionName;
-import com.google.pubsub.v1.ProjectTopicName;
 import com.google.pubsub.v1.Subscription;
 import com.google.pubsub.v1.Topic;
+import com.google.pubsub.v1.TopicName;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.queue.TbQueueAdmin;
 
@@ -103,7 +104,10 @@ public class TbPubSubAdmin implements TbQueueAdmin {
 
     @Override
     public void createTopicIfNotExists(String partition) {
-        ProjectTopicName topicName = ProjectTopicName.of(pubSubSettings.getProjectId(), partition);
+        TopicName topicName = TopicName.newBuilder()
+                .setTopic(partition)
+                .setProject(pubSubSettings.getProjectId())
+                .build();
 
         if (topicSet.contains(topicName.toString())) {
             createSubscriptionIfNotExists(partition, topicName);
@@ -121,13 +125,18 @@ public class TbPubSubAdmin implements TbQueueAdmin {
             }
         }
 
+        try {
             topicAdminClient.createTopic(topicName);
-        topicSet.add(topicName.toString());
             log.info("Created new topic: [{}]", topicName.toString());
+        } catch (AlreadyExistsException e) {
+            log.info("[{}] Topic already exist.", topicName.toString());
+        } finally {
+            topicSet.add(topicName.toString());
+        }
         createSubscriptionIfNotExists(partition, topicName);
     }
 
-    private void createSubscriptionIfNotExists(String partition, ProjectTopicName topicName) {
+    private void createSubscriptionIfNotExists(String partition, TopicName topicName) {
         ProjectSubscriptionName subscriptionName =
                 ProjectSubscriptionName.of(pubSubSettings.getProjectId(), partition);
 
@@ -153,9 +162,14 @@ public class TbPubSubAdmin implements TbQueueAdmin {
         setAckDeadline(subscriptionBuilder);
         setMessageRetention(subscriptionBuilder);
 
+        try {
             subscriptionAdminClient.createSubscription(subscriptionBuilder.build());
-        subscriptionSet.add(subscriptionName.toString());
             log.info("Created new subscription: [{}]", subscriptionName.toString());
+        } catch (AlreadyExistsException e) {
+            log.info("[{}] Subscription already exist.", subscriptionName.toString());
+        } finally {
+            subscriptionSet.add(subscriptionName.toString());
+        }
     }
 
     private void setAckDeadline(Subscription.Builder builder) {

common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubConsumerTemplate.java

@@ -134,6 +134,11 @@ public class TbPubSubConsumerTemplate<T extends TbQueueMsg> implements TbQueueCo
             if (!subscribed) {
                 subscriptionNames = partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.toSet());
                 subscriptionNames.forEach(admin::createTopicIfNotExists);
+
+                if (consumerExecutor != null) {
+                    consumerExecutor.shutdown();
+                }
+
                 consumerExecutor = Executors.newFixedThreadPool(subscriptionNames.size());
                 messagesPerTopic = pubSubSettings.getMaxMessages() / subscriptionNames.size();
                 subscribed = true;

common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubProducerTemplate.java

@@ -124,8 +124,8 @@ public class TbPubSubProducerTemplate<T extends TbQueueMsg> implements TbQueuePr
                 publisherMap.put(topic, publisher);
                 return publisher;
             } catch (IOException e) {
-                log.error("Failed to create topic [{}].", topic, e);
+                log.error("Failed to create Publisher for the topic [{}].", topic, e);
-                throw new RuntimeException("Failed to create topic.", e);
+                throw new RuntimeException("Failed to create Publisher for the topic.", e);
             }
         }
 

msa/js-executor/queue/pubSubTemplate.js

@@ -59,13 +59,11 @@ function PubSubProducer() {
 
         const topicList = await pubSubClient.getTopics();
 
-        if (topicList) {
         if (topicList) {
             topicList[0].forEach(topic => {
                 topics.push(getName(topic.name));
             });
         }
-        }
 
         const subscriptionList = await pubSubClient.getSubscriptions();
 
@@ -100,23 +98,32 @@ function PubSubProducer() {
 
 async function createTopic(topic) {
     if (!topics.includes(topic)) {
+        try {
             await pubSubClient.createTopic(topic);
-        topics.push(topic);
             logger.info('Created new Pub/Sub topic: %s', topic);
+        } catch (e) {
+            logger.info('Pub/Sub topic already exists');
+        }
+        topics.push(topic);
     }
     await createSubscription(topic)
 }
 
 async function createSubscription(topic) {
     if (!subscriptions.includes(topic)) {
+        try {
             await pubSubClient.createSubscription(topic, topic, {
                 topic: topic,
                 subscription: topic,
                 ackDeadlineSeconds: queueProps['ackDeadlineInSec'],
                 messageRetentionDuration: {seconds: queueProps['messageRetentionInSec']}
             });
-        subscriptions.push(topic);
             logger.info('Created new Pub/Sub subscription: %s', topic);
+        } catch (e) {
+            logger.info('Pub/Sub subscription already exists.');
+        }
+
+        subscriptions.push(topic);
     }
 }
 

pom.xml

@@ -95,7 +95,7 @@
         <snakeyaml.version>1.25</snakeyaml.version>
         <struts.version>1.3.10</struts.version>
         <amazonaws.sqs.version>1.11.747</amazonaws.sqs.version>
-        <pubsub.client.version>1.84.0</pubsub.client.version>
+        <pubsub.client.version>1.105.0</pubsub.client.version>
         <azure-servicebus.version>3.2.0</azure-servicebus.version>
         <passay.version>1.5.0</passay.version>
         <ua-parser.version>1.4.3</ua-parser.version>