Merge pull request #9619 from dashevchenko/pubsub

Pubsub executor thread pool size fix

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.PubSubRuleNodeExecutorProvider;
 import org.thingsboard.rule.engine.api.MailService;
 import org.thingsboard.rule.engine.api.NotificationCenter;
 import org.thingsboard.rule.engine.api.SmsService;
@@ -322,6 +323,11 @@ public class ActorSystemContext {
     @Getter
     private NotificationExecutorService notificationExecutor;
 
+    @Lazy
+    @Autowired
+    @Getter
+    private PubSubRuleNodeExecutorProvider pubSubRuleNodeExecutorProvider;
+
     @Autowired
     @Getter
     private SharedEventLoopGroupService sharedEventLoopGroupService;

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

@@ -104,6 +104,7 @@ import org.thingsboard.server.dao.widget.WidgetsBundleService;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.queue.TbQueueCallback;
 import org.thingsboard.server.queue.TbQueueMsgMetadata;
+import org.thingsboard.server.service.executors.PubSubRuleNodeExecutorProvider;
 import org.thingsboard.server.service.script.RuleNodeJsScriptEngine;
 import org.thingsboard.server.service.script.RuleNodeTbelScriptEngine;
 
@@ -538,6 +539,11 @@ class DefaultTbContext implements TbContext {
         return mainCtx.getNotificationExecutor();
     }
 
+    @Override
+    public PubSubRuleNodeExecutorProvider getPubSubRuleNodeExecutorProvider() {
+        return mainCtx.getPubSubRuleNodeExecutorProvider();
+    }
+
     @Override
     @Deprecated
     public ScriptEngine createJsScriptEngine(String script, String... argNames) {

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

@@ -0,0 +1,63 @@
+/**
+ * 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.
+ */
+package org.thingsboard.server.service.executors;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.stereotype.Component;
+import org.thingsboard.common.util.ExecutorProvider;
+import org.thingsboard.common.util.ThingsBoardThreadFactory;
+import org.thingsboard.server.queue.util.TbRuleEngineComponent;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
+@Lazy
+@TbRuleEngineComponent
+@Component
+public class PubSubRuleNodeExecutorProvider implements ExecutorProvider {
+
+    @Value("${service.rule_engine.pubsub.executor_thread_pool_size}")
+    private Integer threadPoolSize;
+
+    /**
+    * Refers to com.google.cloud.pubsub.v1.Publisher default executor configuration
+    */
+    private static final int THREADS_PER_CPU = 5;
+    private ScheduledExecutorService executor;
+
+    @PostConstruct
+    public void init() {
+        if (threadPoolSize == null) {
+            threadPoolSize = THREADS_PER_CPU * Runtime.getRuntime().availableProcessors();
+        }
+        executor = Executors.newScheduledThreadPool(threadPoolSize, ThingsBoardThreadFactory.forName("pubsub-rule-nodes"));
+    }
+
+    @Override
+    public ScheduledExecutorService getExecutor() {
+        return executor;
+    }
+
+    @PreDestroy
+    private void destroy() {
+        if (executor != null) {
+            executor.shutdownNow();
+        }
+    }
+}

application/src/main/resources/thingsboard.yml

@@ -1414,6 +1414,8 @@ queue:
     max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
     # Number of messages per consumer
     max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
+    # Thread pool size for pubsub queue executor provider. If set to 0 - default pubsub executor provider value will be used (5 * number of available processors)
+    executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:0}"
     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}"
@@ -1642,6 +1644,9 @@ service:
     # Comma-separated list of tenant profile ids assigned to this Rule Engine.
     # This Rule Engine will only be responsible for tenants with these profiles (in case 'isolation' option is enabled in the profile).
     assigned_tenant_profiles: "${TB_RULE_ENGINE_ASSIGNED_TENANT_PROFILES:}"
+    pubsub:
+      # Thread pool size for pubsub rule node executor provider. If not set - default pubsub executor provider value will be used (5 * number of available processors)
+      executor_thread_pool_size: "${TB_RULE_ENGINE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:0}"
 
 # Metrics parameters
 metrics:

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

@@ -17,6 +17,7 @@ package org.thingsboard.server.queue.pubsub;
 
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
+import com.google.api.gax.core.FixedExecutorProvider;
 import com.google.cloud.pubsub.v1.stub.GrpcSubscriberStub;
 import com.google.cloud.pubsub.v1.stub.SubscriberStub;
 import com.google.cloud.pubsub.v1.stub.SubscriberStubSettings;
@@ -76,6 +77,7 @@ public class TbPubSubConsumerTemplate<T extends TbQueueMsg> extends AbstractPara
                                     SubscriberStubSettings.defaultGrpcTransportProviderBuilder()
                                             .setMaxInboundMessageSize(pubSubSettings.getMaxMsgSize())
                                             .build())
+                            .setExecutorProvider(pubSubSettings.getExecutorProvider())
                             .build();
             this.subscriber = GrpcSubscriberStub.create(subscriberStubSettings);
         } catch (IOException e) {

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

@@ -18,6 +18,7 @@ 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.FixedExecutorProvider;
 import com.google.cloud.pubsub.v1.Publisher;
 import com.google.gson.Gson;
 import com.google.protobuf.ByteString;
@@ -120,7 +121,10 @@ public class TbPubSubProducerTemplate<T extends TbQueueMsg> implements TbQueuePr
             try {
                 admin.createTopicIfNotExists(topic);
                 ProjectTopicName topicName = ProjectTopicName.of(pubSubSettings.getProjectId(), topic);
-                Publisher publisher = Publisher.newBuilder(topicName).setCredentialsProvider(pubSubSettings.getCredentialsProvider()).build();
+                Publisher publisher = Publisher.newBuilder(topicName)
+                        .setCredentialsProvider(pubSubSettings.getCredentialsProvider())
+                        .setExecutorProvider(pubSubSettings.getExecutorProvider())
+                        .build();
                 publisherMap.put(topic, publisher);
                 return publisher;
             } catch (IOException e) {

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

@@ -17,16 +17,20 @@ package org.thingsboard.server.queue.pubsub;
 
 import com.google.api.gax.core.CredentialsProvider;
 import com.google.api.gax.core.FixedCredentialsProvider;
+import com.google.api.gax.core.FixedExecutorProvider;
 import com.google.auth.oauth2.ServiceAccountCredentials;
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.stereotype.Component;
+import org.thingsboard.common.util.ThingsBoardThreadFactory;
 
 import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.util.concurrent.Executors;
 
 @Slf4j
 @ConditionalOnExpression("'${queue.type:null}'=='pubsub'")
@@ -46,6 +50,16 @@ public class TbPubSubSettings {
     @Value("${queue.pubsub.max_messages}")
     private int maxMessages;
 
+    @Value("${queue.pubsub.executor_thread_pool_size:0}")
+    private int threadPoolSize;
+
+    /**
+     * Refers to com.google.cloud.pubsub.v1.Publisher default executor configuration
+     */
+    private static final int THREADS_PER_CPU = 5;
+
+    private FixedExecutorProvider executorProvider;
+
     private CredentialsProvider credentialsProvider;
 
     @PostConstruct
@@ -53,6 +67,17 @@ public class TbPubSubSettings {
         ServiceAccountCredentials credentials = ServiceAccountCredentials.fromStream(
                 new ByteArrayInputStream(serviceAccount.getBytes()));
         credentialsProvider = FixedCredentialsProvider.create(credentials);
+        if (threadPoolSize == 0) {
+            threadPoolSize = THREADS_PER_CPU * Runtime.getRuntime().availableProcessors();
+        }
+        executorProvider = FixedExecutorProvider
+                .create(Executors.newScheduledThreadPool(threadPoolSize, ThingsBoardThreadFactory.forName("pubsub-queue-executor")));
     }
 
+    @PreDestroy
+    private void destroy() {
+        if (executorProvider != null) {
+            executorProvider.getExecutor().shutdownNow();
+        }
+    }
 }

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

@@ -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.
+ */
+package org.thingsboard.common.util;
+
+import java.util.concurrent.ScheduledExecutorService;
+
+public interface ExecutorProvider {
+
+    ScheduledExecutorService getExecutor();
+}

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

@@ -172,6 +172,8 @@ 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}"
+    # Thread pool size for pubsub queue executor provider. If not set - default pubsub executor provider value will be used (5 * number of available processors)
+    executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:0}"
     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

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

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

@@ -20,6 +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.FixedExecutorProvider;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.auth.oauth2.ServiceAccountCredentials;
 import com.google.cloud.pubsub.v1.Publisher;
@@ -68,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);
         } catch (Exception e) {
             throw new TbNodeException(e);
         }
@@ -128,7 +129,7 @@ public class TbPubSubNode extends TbAbstractExternalNode {
         return TbMsg.transformMsgMetadata(origMsg, metaData);
     }
 
-    private Publisher initPubSubClient() throws IOException {
+    private Publisher initPubSubClient(TbContext ctx) throws IOException {
         ProjectTopicName topicName = ProjectTopicName.of(config.getProjectId(), config.getTopicName());
         ServiceAccountCredentials credentials =
                 ServiceAccountCredentials.fromStream(
@@ -148,6 +149,7 @@ public class TbPubSubNode extends TbAbstractExternalNode {
         return Publisher.newBuilder(topicName)
                 .setCredentialsProvider(credProvider)
                 .setRetrySettings(retrySettings)
+                .setExecutorProvider(FixedExecutorProvider.create(ctx.getPubSubRuleNodeExecutorProvider().getExecutor()))
                 .build();
     }
 }

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

@@ -299,6 +299,8 @@ queue:
     max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
     # Number of messages per consumer
     max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
+    # Thread pool size for pubsub queue executor provider. If not set - default pubsub executor provider value will be used (5 * number of available processors)
+    executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:0}"
     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

@@ -282,6 +282,8 @@ 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}"
+    # Thread pool size for pubsub queue executor provider. If not set - default pubsub executor provider value will be used (5 * number of available processors)
+    executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:0}"
     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

@@ -378,6 +378,8 @@ queue:
     max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
     # Number of messages per consumer
     max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
+    # Thread pool size for pubsub queue executor provider. If not set - default pubsub executor provider value will be used (5 * number of available processors)
+    executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:0}"
     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

@@ -315,6 +315,8 @@ queue:
     max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
     # Number of messages per consumer
     max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
+    # Thread pool size for pubsub queue executor provider. If not set - default pubsub executor provider value will be used (5 * number of available processors)
+    executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:0}"
     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

@@ -268,6 +268,8 @@ queue:
     max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
     # Number of messages per consumer
     max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
+    # Thread pool size for pubsub queue executor provider. If not set - default pubsub executor provider value will be used (5 * number of available processors)
+    executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:0}"
     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}"