From 6a706b32da934eac55d288aacba1e3a231f458b2 Mon Sep 17 00:00:00 2001 From: Sergey Matvienko Date: Tue, 24 Aug 2021 19:11:37 +0300 Subject: [PATCH] executors: shutdownNow added for tb-rule-engine-consumer-repartition executor on @PreDestroy --- .../service/queue/DefaultTbRuleEngineConsumerService.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java index 4b1392e14a..953d45d83d 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java @@ -103,8 +103,8 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< private final ConcurrentMap consumerConfigurations = new ConcurrentHashMap<>(); private final ConcurrentMap consumerStats = new ConcurrentHashMap<>(); private final ConcurrentMap topicsConsumerPerPartition = new ConcurrentHashMap<>(); - final ExecutorService submitExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer-service-submit-executor")); - final ScheduledExecutorService repartitionExecutor = Executors.newScheduledThreadPool(1, ThingsBoardThreadFactory.forName("tb-rule-engine-consumer-repartition-executor")); + final ExecutorService submitExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer-submit")); + final ScheduledExecutorService repartitionExecutor = Executors.newScheduledThreadPool(1, ThingsBoardThreadFactory.forName("tb-rule-engine-consumer-repartition")); public DefaultTbRuleEngineConsumerService(TbRuleEngineProcessingStrategyFactory processingStrategyFactory, TbRuleEngineSubmitStrategyFactory submitStrategyFactory, @@ -146,6 +146,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< public void stop() { super.destroy(); submitExecutor.shutdownNow(); + repartitionExecutor.shutdownNow(); ruleEngineSettings.getQueues().forEach(config -> consumerConfigurations.put(config.getName(), config)); }