implemented PubSubExecutor to use in all pubsub rule nodes

application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java

@@ -29,6 +29,7 @@ import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.server.service.executors.PubSubExecutorService;
 import org.thingsboard.rule.engine.api.MailService;
 import org.thingsboard.rule.engine.api.NotificationCenter;
 import org.thingsboard.rule.engine.api.SmsService;
@@ -322,6 +323,10 @@ public class ActorSystemContext {
     @Getter
     private NotificationExecutorService notificationExecutor;
 
+    @Autowired
+    @Getter
+    private PubSubExecutorService pubSubExecutorService;
+
     @Autowired
     @Getter
     private SharedEventLoopGroupService sharedEventLoopGroupService;

application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java

@@ -22,6 +22,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.bouncycastle.util.Arrays;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.common.util.ListeningExecutor;
+import org.thingsboard.server.service.executors.PubSubExecutorService;
 import org.thingsboard.rule.engine.api.MailService;
 import org.thingsboard.rule.engine.api.NotificationCenter;
 import org.thingsboard.rule.engine.api.RuleEngineAlarmService;
@@ -538,6 +539,11 @@ class DefaultTbContext implements TbContext {
         return mainCtx.getNotificationExecutor();
     }
 
+    @Override
+    public PubSubExecutorService getPubsubExecutor() {
+        return mainCtx.getPubSubExecutorService();
+    }
+
     @Override
     @Deprecated
     public ScriptEngine createJsScriptEngine(String script, String... argNames) {

application/src/main/java/org/thingsboard/server/service/executors/PubSubExecutorService.java

@@ -0,0 +1,26 @@
+package org.thingsboard.server.service.executors;
+
+import com.google.api.gax.core.FixedExecutorProvider;
+import org.springframework.stereotype.Component;
+import org.thingsboard.common.util.PubSubExecutor;
+import org.thingsboard.common.util.ThingsBoardThreadFactory;
+
+import javax.annotation.PostConstruct;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
+@Component
+public class PubSubExecutorService implements PubSubExecutor {
+    private static final int THREADS_PER_CPU = 5;
+    private FixedExecutorProvider fixedExecutorProvider;
+
+    @PostConstruct
+    public void init() {
+        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(THREADS_PER_CPU * Runtime.getRuntime().availableProcessors(), ThingsBoardThreadFactory.forName("tb-pubsub-producer-scheduler"));;
+        this.fixedExecutorProvider = FixedExecutorProvider.create(scheduler);
+    }
+
+    public FixedExecutorProvider getExecutorProvider() {
+        return fixedExecutorProvider;
+    }
+}

application/src/main/resources/thingsboard.yml

@@ -1411,8 +1411,6 @@ queue:
     max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
     # Number of messages per consumer
     max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
-    # Number of threads of pubsub executor provider
-    executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:50}"
     queue-properties:
       # Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consumed again
       rule-engine: "${TB_QUEUE_PUBSUB_RE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}"

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

@@ -18,13 +18,14 @@ package org.thingsboard.server.queue.pubsub;
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutureCallback;
 import com.google.api.core.ApiFutures;
-import com.google.api.gax.core.InstantiatingExecutorProvider;
+import com.google.api.gax.core.FixedExecutorProvider;
 import com.google.cloud.pubsub.v1.Publisher;
 import com.google.gson.Gson;
 import com.google.protobuf.ByteString;
 import com.google.pubsub.v1.ProjectTopicName;
 import com.google.pubsub.v1.PubsubMessage;
 import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.queue.TbQueueAdmin;
 import org.thingsboard.server.queue.TbQueueCallback;
@@ -37,6 +38,7 @@ import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
 @Slf4j
@@ -51,11 +53,16 @@ public class TbPubSubProducerTemplate<T extends TbQueueMsg> implements TbQueuePr
     private final Map<String, Publisher> publisherMap = new ConcurrentHashMap<>();
 
     private final ExecutorService pubExecutor = Executors.newCachedThreadPool();
+    private static final int THREADS_PER_CPU = 5;
+    private final FixedExecutorProvider fixedExecutorProvider;
 
     public TbPubSubProducerTemplate(TbQueueAdmin admin, TbPubSubSettings pubSubSettings, String defaultTopic) {
         this.defaultTopic = defaultTopic;
         this.admin = admin;
         this.pubSubSettings = pubSubSettings;
+
+        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(THREADS_PER_CPU * Runtime.getRuntime().availableProcessors(), ThingsBoardThreadFactory.forName("tb-pubsub-producer-scheduler"));;
+        fixedExecutorProvider = FixedExecutorProvider.create(scheduler);
     }
 
     @Override
@@ -121,13 +128,9 @@ public class TbPubSubProducerTemplate<T extends TbQueueMsg> implements TbQueuePr
             try {
                 admin.createTopicIfNotExists(topic);
                 ProjectTopicName topicName = ProjectTopicName.of(pubSubSettings.getProjectId(), topic);
-                InstantiatingExecutorProvider executorProvider =
-                        InstantiatingExecutorProvider.newBuilder()
-                                .setExecutorThreadCount(pubSubSettings.getExecutorThreadPoolSize())
-                                .build();
                 Publisher publisher = Publisher.newBuilder(topicName)
                         .setCredentialsProvider(pubSubSettings.getCredentialsProvider())
-                        .setExecutorProvider(executorProvider)
+                        .setExecutorProvider(fixedExecutorProvider)
                         .build();
                 publisherMap.put(topic, publisher);
                 return publisher;

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

@@ -46,9 +46,6 @@ public class TbPubSubSettings {
     @Value("${queue.pubsub.max_messages}")
     private int maxMessages;
 
-    @Value("${queue.pubsub.executor_thread_pool_size}")
-    private int executorThreadPoolSize;
-
     private CredentialsProvider credentialsProvider;
 
     @PostConstruct

common/util/pom.xml

@@ -53,6 +53,10 @@
             <artifactId>guava</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>com.google.cloud</groupId>
+            <artifactId>google-cloud-pubsub</artifactId>
+        </dependency>
         <dependency>
             <groupId>javax.annotation</groupId>
             <artifactId>javax.annotation-api</artifactId>

common/util/src/main/java/org/thingsboard/common/util/PubSubExecutor.java

@@ -0,0 +1,7 @@
+package org.thingsboard.common.util;
+
+import com.google.api.gax.core.FixedExecutorProvider;
+
+public interface PubSubExecutor {
+    FixedExecutorProvider getExecutorProvider();
+}

msa/vc-executor/src/main/resources/tb-vc-executor.yml

@@ -172,8 +172,6 @@ queue:
     max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}" #in bytes
     # Number of messages per consumer
     max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
-    # Number of threads of pubsub executor provider
-    executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:50}"
     queue-properties:
       # Pub/Sub properties for Core subscribers, messages which will commit after ackDeadlineInSec period can be consumed again
       core: "${TB_QUEUE_PUBSUB_CORE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}"

rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java

@@ -17,6 +17,7 @@ package org.thingsboard.rule.engine.api;
 
 import io.netty.channel.EventLoopGroup;
 import org.thingsboard.common.util.ListeningExecutor;
+import org.thingsboard.common.util.PubSubExecutor;
 import org.thingsboard.rule.engine.api.slack.SlackService;
 import org.thingsboard.rule.engine.api.sms.SmsSenderFactory;
 import org.thingsboard.server.cluster.TbClusterService;
@@ -318,6 +319,8 @@ public interface TbContext {
 
     ListeningExecutor getNotificationExecutor();
 
+    PubSubExecutor getPubsubExecutor();
+
     MailService getMailService(boolean isSystem);
 
     SmsService getSmsService();

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/gcp/pubsub/TbPubSubNode.java

@@ -20,7 +20,7 @@ import com.google.api.core.ApiFutureCallback;
 import com.google.api.core.ApiFutures;
 import com.google.api.gax.core.CredentialsProvider;
 import com.google.api.gax.core.FixedCredentialsProvider;
-import com.google.api.gax.core.InstantiatingExecutorProvider;
+import com.google.api.gax.core.FixedExecutorProvider;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.auth.oauth2.ServiceAccountCredentials;
 import com.google.cloud.pubsub.v1.Publisher;
@@ -69,7 +69,7 @@ public class TbPubSubNode extends TbAbstractExternalNode {
         super.init(ctx);
         this.config = TbNodeUtils.convert(configuration, TbPubSubNodeConfiguration.class);
         try {
-            this.pubSubClient = initPubSubClient();
+            this.pubSubClient = initPubSubClient(ctx.getPubsubExecutor().getExecutorProvider());
         } catch (Exception e) {
             throw new TbNodeException(e);
         }
@@ -129,18 +129,13 @@ public class TbPubSubNode extends TbAbstractExternalNode {
         return TbMsg.transformMsgMetadata(origMsg, metaData);
     }
 
-    private Publisher initPubSubClient() throws IOException {
+    private Publisher initPubSubClient(FixedExecutorProvider fixedExecutorProvider) throws IOException {
         ProjectTopicName topicName = ProjectTopicName.of(config.getProjectId(), config.getTopicName());
         ServiceAccountCredentials credentials =
                 ServiceAccountCredentials.fromStream(
                         new ByteArrayInputStream(config.getServiceAccountKey().getBytes()));
         CredentialsProvider credProvider = FixedCredentialsProvider.create(credentials);
 
-        InstantiatingExecutorProvider executorProvider =
-                InstantiatingExecutorProvider.newBuilder()
-                        .setExecutorThreadCount(1)
-                        .build();
-
         var retrySettings = RetrySettings.newBuilder()
                 .setTotalTimeout(Duration.ofSeconds(10))
                 .setInitialRetryDelay(Duration.ofMillis(50))
@@ -154,7 +149,7 @@ public class TbPubSubNode extends TbAbstractExternalNode {
         return Publisher.newBuilder(topicName)
                 .setCredentialsProvider(credProvider)
                 .setRetrySettings(retrySettings)
-                .setExecutorProvider(executorProvider)
+                .setExecutorProvider(fixedExecutorProvider)
                 .build();
     }
 }

transport/coap/src/main/resources/tb-coap-transport.yml

@@ -295,8 +295,6 @@ queue:
     max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
     # Number of messages per consumer
     max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
-    # Number of threads of pubsub executor provider
-    executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:50}"
     queue-properties:
       # Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consumed again
       rule-engine: "${TB_QUEUE_PUBSUB_RE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}"

transport/http/src/main/resources/tb-http-transport.yml

@@ -278,8 +278,6 @@ queue:
     max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
     # Number of messages per a consumer
     max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
-    # Number of threads of pubsub executor provider
-    executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:50}"
     queue-properties:
       # Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consume again
       rule-engine: "${TB_QUEUE_PUBSUB_RE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}"

transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml

@@ -374,8 +374,6 @@ queue:
     max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
     # Number of messages per consumer
     max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
-    # Number of threads of pubsub executor provider
-    executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:50}"
     queue-properties:
       # Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consumed again
       rule-engine: "${TB_QUEUE_PUBSUB_RE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}"

transport/mqtt/src/main/resources/tb-mqtt-transport.yml

@@ -311,8 +311,6 @@ queue:
     max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
     # Number of messages per consumer
     max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
-    # Number of threads of pubsub executor provider
-    executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:50}"
     queue-properties:
       # Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consumed again
       rule-engine: "${TB_QUEUE_PUBSUB_RE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}"

transport/snmp/src/main/resources/tb-snmp-transport.yml

@@ -264,8 +264,6 @@ queue:
     max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
     # Number of messages per consumer
     max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
-    # Number of threads of pubsub executor provider
-    executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:50}"
     queue-properties:
       # Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consumed again
       rule-engine: "${TB_QUEUE_PUBSUB_RE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}"