From 055919c8a7efa1238deb2a48086e64f43c927d59 Mon Sep 17 00:00:00 2001
From: Volodymyr Babak <volodymyr.babak@gmail.com>
Date: Tue, 28 Jun 2022 12:33:45 +0300
Subject: [PATCH] Saving of edge event in batches should be single threaded.
 persisten attributes should not be propagated to edge

---
 .../service/edge/rpc/sync/DefaultEdgeRequestsService.java    | 4 ++++
 application/src/main/resources/thingsboard.yml               | 1 -
 .../thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java | 5 +----
 3 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java
index 210ad30c44..c42de1b68c 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java
@@ -75,6 +75,7 @@ import org.thingsboard.server.gen.edge.v1.UserCredentialsRequestMsg;
 import org.thingsboard.server.gen.edge.v1.WidgetBundleTypesRequestMsg;
 import org.thingsboard.server.service.entitiy.entityView.TbEntityViewService;
 import org.thingsboard.server.service.executors.DbCallbackExecutorService;
+import org.thingsboard.server.service.state.DefaultDeviceStateService;
 
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -163,6 +164,9 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService {
                     Map<String, Object> entityData = new HashMap<>();
                     ObjectNode attributes = mapper.createObjectNode();
                     for (AttributeKvEntry attr : ssAttributes) {
+                        if (DefaultDeviceStateService.PERSISTENT_ATTRIBUTES.contains(attr.getKey())) {
+                            continue;
+                        }
                         if (attr.getDataType() == DataType.BOOLEAN && attr.getBooleanValue().isPresent()) {
                             attributes.put(attr.getKey(), attr.getBooleanValue().get());
                         } else if (attr.getDataType() == DataType.DOUBLE && attr.getDoubleValue().isPresent()) {
diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml
index f6312a7659..05cd5a2599 100644
--- a/application/src/main/resources/thingsboard.yml
+++ b/application/src/main/resources/thingsboard.yml
@@ -266,7 +266,6 @@ sql:
     batch_size: "${SQL_EDGE_EVENTS_BATCH_SIZE:1000}"
     batch_max_delay: "${SQL_EDGE_EVENTS_BATCH_MAX_DELAY_MS:100}"
     stats_print_interval_ms: "${SQL_EDGE_EVENTS_BATCH_STATS_PRINT_MS:10000}"
-    batch_threads: "${SQL_EDGE_EVENTS_BATCH_THREADS:3}" # batch thread count have to be a prime number like 3 or 5 to gain perfect hash distribution
   # Specify whether to sort entities before batch update. Should be enabled for cluster mode to avoid deadlocks
   batch_sort: "${SQL_BATCH_SORT:false}"
   # Specify whether to remove null characters from strValue of attributes and timeseries before insert
diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java
index 5ecb60e744..36f834b8bc 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java
@@ -72,9 +72,6 @@ public class JpaBaseEdgeEventDao extends JpaAbstractSearchTextDao<EdgeEventEntit
     @Value("${sql.edge_events.stats_print_interval_ms:10000}")
     private long statsPrintIntervalMs;
 
-    @Value("${sql.edge_events.batch_threads:3}")
-    private int batchThreads;
-
     private TbSqlBlockingQueueWrapper<EdgeEventEntity> queue;
 
     @Autowired
@@ -110,7 +107,7 @@ public class JpaBaseEdgeEventDao extends JpaAbstractSearchTextDao<EdgeEventEntit
                 return NULL_UUID.hashCode();
             }
         };
-        queue = new TbSqlBlockingQueueWrapper<>(params, hashcodeFunction, batchThreads, statsFactory);
+        queue = new TbSqlBlockingQueueWrapper<>(params, hashcodeFunction, 1, statsFactory);
         queue.init(logExecutor, v -> edgeEventInsertRepository.save(v),
                 Comparator.comparing(EdgeEventEntity::getTs)
         );