refactored

dao/src/main/java/org/thingsboard/server/dao/service/DataValidator.java

@@ -64,7 +64,7 @@ public abstract class DataValidator<D extends BaseData<?>> {
         return actualData.getId() != null && existentData.getId().equals(actualData.getId());
     }
 
-    public static void validateEmail(String email) {
+    protected static void validateEmail(String email) {
         if (!doValidateEmail(email)) {
             throw new DataValidationException("Invalid email address format '" + email + "'!");
         }

dao/src/main/java/org/thingsboard/server/dao/sql/TbSqlBlockingQueueWrapper.java

@@ -42,8 +42,7 @@ public class TbSqlBlockingQueueWrapper<E> {
     }
 
     public ListenableFuture<Void> add(E element) {
-        int hash = hashCodeFunction.apply(element);
-        int queueIndex = (hash & 0x7FFFFFFF) % maxThreads;
+        int queueIndex = element != null ? (hashCodeFunction.apply(element) & 0x7FFFFFFF) % maxThreads : 0;
         return queues.get(queueIndex).add(element);
     }
 

dao/src/main/java/org/thingsboard/server/dao/sql/attributes/JpaAttributeDao.java

@@ -83,7 +83,7 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl
                 .statsPrintIntervalMs(statsPrintIntervalMs)
                 .build();
 
-        Function<AttributeKvEntity, Integer> hashcodeFunction = entity -> entity != null ? entity.getId().getEntityId().hashCode() : 0;
+        Function<AttributeKvEntity, Integer> hashcodeFunction = entity -> entity.getId().getEntityId().hashCode();
         queue = new TbSqlBlockingQueueWrapper<>(params, hashcodeFunction, batchThreads);
         queue.init(logExecutor, v -> attributeKvInsertRepository.saveOrUpdate(v));
     }

dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractChunkedAggregationTimeseriesDao.java

@@ -68,7 +68,7 @@ public abstract class AbstractChunkedAggregationTimeseriesDao extends AbstractSq
                 .statsPrintIntervalMs(tsStatsPrintIntervalMs)
                 .build();
 
-        Function<TsKvEntity, Integer> hashcodeFunction = entity -> entity != null ? entity.getEntityId().hashCode() : 0;
+        Function<TsKvEntity, Integer> hashcodeFunction = entity -> entity.getEntityId().hashCode();
         tsQueue = new TbSqlBlockingQueueWrapper<>(tsParams, hashcodeFunction, tsBatchThreads);
         tsQueue.init(logExecutor, v -> insertRepository.saveOrUpdate(v));
     }

dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractSqlTimeseriesDao.java

@@ -125,7 +125,7 @@ public abstract class AbstractSqlTimeseriesDao extends JpaAbstractDaoListeningEx
                 .statsPrintIntervalMs(tsLatestStatsPrintIntervalMs)
                 .build();
 
-        java.util.function.Function<TsKvLatestEntity, Integer> hashcodeFunction = entity -> entity != null ? entity.getEntityId().hashCode() : 0;
+        java.util.function.Function<TsKvLatestEntity, Integer> hashcodeFunction = entity -> entity.getEntityId().hashCode();
         tsLatestQueue = new TbSqlBlockingQueueWrapper<>(tsLatestParams, hashcodeFunction, tsLatestBatchThreads);
 
         tsLatestQueue.init(logExecutor, v -> {

dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TimescaleTimeseriesDao.java

@@ -76,7 +76,7 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements
                 .statsPrintIntervalMs(tsStatsPrintIntervalMs)
                 .build();
 
-        Function<TimescaleTsKvEntity, Integer> hashcodeFunction = entity -> entity != null ? entity.getEntityId().hashCode() : 0;
+        Function<TimescaleTsKvEntity, Integer> hashcodeFunction = entity -> entity.getEntityId().hashCode();
         tsQueue = new TbSqlBlockingQueueWrapper<>(tsParams, hashcodeFunction, timescaleBatchThreads);
 
         tsQueue.init(logExecutor, v -> insertRepository.saveOrUpdate(v));
@@ -281,4 +281,5 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements
                 startTs,
                 endTs);
     }
+
 }