From 3a6a44a9a4567ed97bada32c182193c7405e792d Mon Sep 17 00:00:00 2001
From: Andrii Landiak <andriylandiak@gmail.com>
Date: Tue, 3 Sep 2024 13:42:28 +0300
Subject: [PATCH] Refactoring after review

---
 .../service/queue/DefaultTbCoreConsumerService.java  |  2 +-
 .../service/queue/DefaultTbEdgeConsumerService.java  | 12 +++++-------
 .../server/service/queue/TbEdgeConsumerService.java  |  6 +-----
 application/src/main/resources/thingsboard.yml       |  2 ++
 4 files changed, 9 insertions(+), 13 deletions(-)

diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
index d30f03262d..2a64dd3388 100644
--- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
+++ b/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()
diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java
index 6ec5f303fc..805aeb0bdb 100644
--- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java
+++ b/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);
         }
 
diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbEdgeConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/TbEdgeConsumerService.java
index 6eaac782e0..086a4b2a63 100644
--- a/application/src/main/java/org/thingsboard/server/service/queue/TbEdgeConsumerService.java
+++ b/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 {
 }
diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml
index 9de3d88568..aefb78106d 100644
--- a/application/src/main/resources/thingsboard.yml
+++ b/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}"