Refactoring after review

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

@@ -243,7 +243,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
 
     @Override
     protected void onTbApplicationEvent(PartitionChangeEvent event) {
-        log.info("Subscribing to partitions: {}", event.getCorePartitions());
+        log.debug("Subscribing to partitions: {}", event.getCorePartitions());
         mainConsumer.update(event.getCorePartitions());
         usageStatsConsumer.subscribe(event.getCorePartitions()
                 .stream()

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

@@ -76,7 +76,7 @@ public class DefaultTbEdgeConsumerService extends AbstractConsumerService<ToEdge
     private int pollInterval;
     @Value("${queue.edge.pack-processing-timeout:10000}")
     private int packProcessingTimeout;
-    @Value("${queue.core.consumer-per-partition:true}")
+    @Value("${queue.edge.consumer-per-partition:false}")
     private boolean consumerPerPartition;
     @Value("${queue.edge.pack-processing-retries:3}")
     private int packProcessingRetries;
@@ -125,11 +125,9 @@ public class DefaultTbEdgeConsumerService extends AbstractConsumerService<ToEdge
 
     @Override
     protected void onTbApplicationEvent(PartitionChangeEvent event) {
-        if (ServiceType.TB_CORE.equals(event.getServiceType())) {
-            var partitions = event.getEdgePartitions();
-            log.info("Subscribing to partitions: {}", partitions);
-            mainConsumer.update(partitions);
-        }
+        var partitions = event.getEdgePartitions();
+        log.debug("Subscribing to partitions: {}", partitions);
+        mainConsumer.update(partitions);
     }
 
     private void processMsgs(List<TbProtoQueueMsg<ToEdgeMsg>> msgs, TbQueueConsumer<TbProtoQueueMsg<ToEdgeMsg>> consumer, EdgeQueueConfig edgeQueueConfig) throws InterruptedException {
@@ -235,7 +233,7 @@ public class DefaultTbEdgeConsumerService extends AbstractConsumerService<ToEdge
                 callback.onSuccess();
             }
         } catch (Exception e) {
-            log.error("Error processing edge notification message: {}", e.getMessage(), e);
+            log.error("Error processing edge notification message", e);
             callback.onFailure(e);
         }
 

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

@@ -15,9 +15,5 @@
  */
 package org.thingsboard.server.service.queue;
 
-import org.springframework.context.ApplicationListener;
-import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
-
-public interface TbEdgeConsumerService extends ApplicationListener<PartitionChangeEvent> {
-
+public interface TbEdgeConsumerService {
 }

application/src/main/resources/thingsboard.yml

@@ -1798,6 +1798,8 @@ queue:
     pack-processing-timeout: "${TB_QUEUE_EDGE_PACK_PROCESSING_TIMEOUT_MS:10000}"
     # Retries for processing a failure message pack by Edge services
     pack-processing-retries: "${TB_QUEUE_EDGE_MESSAGE_PROCESSING_RETRIES:3}"
+    # Enable/disable a separate consumer per partition for Edge queue
+    consumer-per-partition: "${TB_QUEUE_EDGE_CONSUMER_PER_PARTITION:false}"
     stats:
       # Enable/disable statistics for Edge services
       enabled: "${TB_QUEUE_EDGE_STATS_ENABLED:true}"