Minor refactoring for consumer managers

application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java

@@ -127,8 +127,8 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
     private long pollInterval;
     @Value("${queue.core.pack-processing-timeout}")
     private long packProcessingTimeout;
-    @Value("${queue.core.consumer-per-partition-enabled:true}")
+    @Value("${queue.core.consumer-per-partition:true}")
-    private boolean consumerPerPartitionEnabled;
+    private boolean consumerPerPartition;
     @Value("${queue.core.stats.enabled:false}")
     private boolean statsEnabled;
 
@@ -201,7 +201,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
 
         this.mainConsumer = QueueConsumerManager.<TbProtoQueueMsg<ToCoreMsg>, CoreQueueConfig>builder()
                 .queueKey(new QueueKey(ServiceType.TB_CORE))
-                .config(CoreQueueConfig.of(consumerPerPartitionEnabled, (int) pollInterval))
+                .config(CoreQueueConfig.of(consumerPerPartition, (int) pollInterval))
                 .msgPackProcessor(this::processMsgs)
                 .consumerCreator(config -> queueFactory.createToCoreMsgConsumer())
                 .consumerExecutor(consumersExecutor)
@@ -325,7 +325,9 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
                 ToCoreMsg lastSubmitMsg = pendingMsgHolder.getToCoreMsg();
                 log.info("Timeout to process message: {}", lastSubmitMsg);
             }
-            ctx.getAckMap().forEach((id, msg) -> log.debug("[{}] Timeout to process message: {}", id, msg.getValue()));
+            if (log.isDebugEnabled()) {
+                ctx.getAckMap().forEach((id, msg) -> log.debug("[{}] Timeout to process message: {}", id, msg.getValue()));
+            }
             ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process message: {}", id, msg.getValue()));
         }
         consumer.commit();
@@ -452,6 +454,8 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
                     }
                     timeToSleep += maxProcessingTimeoutPerRecord;
                 }
+            } catch (InterruptedException e) {
+                return;
             } catch (Throwable e) {
                 log.warn("Failed to process firmware update msg: {}", msg, e);
             }

application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java

@@ -15,6 +15,7 @@
  */
 package org.thingsboard.server.service.queue;
 
+import jakarta.annotation.PostConstruct;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.event.EventListener;
@@ -43,9 +44,6 @@ import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.discovery.QueueKey;
 import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
-import org.thingsboard.server.queue.util.DataDecodingEncodingService;
-import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory;
-import org.thingsboard.server.queue.util.AfterStartUp;
 import org.thingsboard.server.queue.util.TbRuleEngineComponent;
 import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
 import org.thingsboard.server.service.profile.TbAssetProfileCache;
@@ -56,7 +54,6 @@ import org.thingsboard.server.service.queue.ruleengine.TbRuleEngineQueueConsumer
 import org.thingsboard.server.service.rpc.TbRuleEngineDeviceRpcService;
 import org.thingsboard.server.service.security.auth.jwt.settings.JwtSettingsService;
 
-import jakarta.annotation.PostConstruct;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;

application/src/main/java/org/thingsboard/server/service/queue/consumer/BasicQueueConsumerManager.java

@@ -72,6 +72,7 @@ public class BasicQueueConsumerManager<M extends TbQueueMsg> {
             } catch (Throwable e) {
                 log.error("Failure in consumer loop", e);
             }
+            log.info("[{}] Consumer stopped", name);
         });
     }
 
@@ -94,7 +95,6 @@ public class BasicQueueConsumerManager<M extends TbQueueMsg> {
                 }
             }
         }
-        log.info("{} Consumer stopped", name);
     }
 
     public void stop() {

application/src/main/java/org/thingsboard/server/service/queue/consumer/QueueConsumerManager.java

@@ -176,8 +176,9 @@ public class QueueConsumerManager<M extends TbQueueMsg, C extends QueueConfig> {
                 doUpdate(partitions); // even if partitions number was changed, there can be no partition change event
             }
         } else {
+            log.trace("[{}] Silently applied new config, because consumer-per-partition not changed", queueKey);
             // do nothing, because partitions change (if they changed) will be handled on PartitionChangeEvent,
-            // and changes to pollInterval/packProcessingTimeout/submitStrategy/processingStrategy will be picked up by consumer on the fly,
+            // and changes to other config values will be picked up by consumer on the fly,
             // and queue topic and name are immutable
         }
     }
@@ -196,6 +197,7 @@ public class QueueConsumerManager<M extends TbQueueMsg, C extends QueueConfig> {
             } catch (Throwable e) {
                 log.error("Failure in consumer loop", e);
             }
+            log.info("[{}] Consumer stopped", consumerTask.getKey());
         });
         consumerTask.setTask(consumerLoop);
     }
@@ -222,7 +224,6 @@ public class QueueConsumerManager<M extends TbQueueMsg, C extends QueueConfig> {
         if (consumer.isStopped()) {
             consumer.unsubscribe();
         }
-        log.info("{} Consumer stopped", queueKey);
     }
 
     protected void processMsgs(List<M> msgs, TbQueueConsumer<M> consumer, C config) throws Exception {
@@ -236,6 +237,7 @@ public class QueueConsumerManager<M extends TbQueueMsg, C extends QueueConfig> {
     }
 
     public void awaitStop() {
+        log.debug("[{}] Waiting for consumers to stop", queueKey);
         consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::awaitCompletion);
         log.debug("[{}] Unsubscribed and stopped consumers", queueKey);
     }
@@ -272,7 +274,7 @@ public class QueueConsumerManager<M extends TbQueueMsg, C extends QueueConfig> {
             removedPartitions.forEach((tpi) -> consumers.remove(tpi).awaitCompletion());
 
             addedPartitions.forEach((tpi) -> {
-                String key = queueKey + "-" + tpi.getPartition().orElse(-999999);
+                String key = queueKey + "-" + tpi.getPartition().orElse(-1);
                 TbQueueConsumerTask<M> consumer = new TbQueueConsumerTask<>(key, consumerCreator.apply(config));
                 consumers.put(tpi, consumer);
                 consumer.subscribe(Set.of(tpi));

application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java

@@ -15,6 +15,7 @@
  */
 package org.thingsboard.server.service.queue.processing;
 
+import jakarta.annotation.PreDestroy;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.ApplicationEventPublisher;
@@ -48,7 +49,6 @@ import org.thingsboard.server.service.queue.TbPackProcessingContext;
 import org.thingsboard.server.service.queue.consumer.BasicQueueConsumerManager;
 import org.thingsboard.server.service.security.auth.jwt.settings.JwtSettingsService;
 
-import jakarta.annotation.PreDestroy;
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;

application/src/main/resources/thingsboard.yml

@@ -1596,7 +1596,7 @@ queue:
     # Timeout for processing a message pack by Core microservices
     pack-processing-timeout: "${TB_QUEUE_CORE_PACK_PROCESSING_TIMEOUT_MS:2000}"
     # Enable/disable a separate consumer per partition for Core queue
-    consumer-per-partition-enabled: "${TB_QUEUE_CORE_CONSUMER_PER_PARTITION_ENABLED:true}"
+    consumer-per-partition: "${TB_QUEUE_CORE_CONSUMER_PER_PARTITION:true}"
     ota:
       # Default topic name for OTA updates
       topic: "${TB_QUEUE_CORE_OTA_TOPIC:tb_ota_package}"

application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java

@@ -453,7 +453,7 @@ public class TbRuleEngineQueueConsumerManagerTest {
         int msgCount = totalConsumedMsgs.get();
 
         await().atLeast(2, TimeUnit.SECONDS) // based on topicDeletionDelayInSec(5) = 5 - ( 3 seconds the code may execute starting consumerManager.delete() call)
-                .atMost(10, TimeUnit.SECONDS)
+                .atMost(20, TimeUnit.SECONDS)
                 .untilAsserted(() -> {
                     partitions.stream()
                             .map(TopicPartitionInfo::getFullTopicName)
@@ -492,7 +492,7 @@ public class TbRuleEngineQueueConsumerManagerTest {
         int msgCount = totalConsumedMsgs.get();
 
         await().atLeast(2, TimeUnit.SECONDS) // based on topicDeletionDelayInSec(5) = 5 - ( 3 seconds the code may execute starting consumerManager.delete() call)
-                .atMost(10, TimeUnit.SECONDS)
+                .atMost(20, TimeUnit.SECONDS)
                 .untilAsserted(() -> {
                     partitions.stream()
                             .map(TopicPartitionInfo::getFullTopicName)