Merge pull request #10337 from smatvienko-tb/feature/latest-ts-async-improvement

Latest ts and attribute async improvement

dao/src/main/java/org/thingsboard/server/dao/attributes/CachedAttributesService.java

@@ -113,15 +113,15 @@ public class CachedAttributesService implements AttributesService {
         validate(entityId, scope);
         Validator.validateString(attributeKey, "Incorrect attribute key " + attributeKey);
 
-        AttributeCacheKey attributeCacheKey = new AttributeCacheKey(scope, entityId, attributeKey);
+        return cacheExecutor.submit(() -> {
-        TbCacheValueWrapper<AttributeKvEntry> cachedAttributeValue = cache.get(attributeCacheKey);
+            AttributeCacheKey attributeCacheKey = new AttributeCacheKey(scope, entityId, attributeKey);
-        if (cachedAttributeValue != null) {
+            TbCacheValueWrapper<AttributeKvEntry> cachedAttributeValue = cache.get(attributeCacheKey);
-            hitCounter.increment();
+            if (cachedAttributeValue != null) {
-            AttributeKvEntry cachedAttributeKvEntry = cachedAttributeValue.get();
+                hitCounter.increment();
-            return Futures.immediateFuture(Optional.ofNullable(cachedAttributeKvEntry));
+                AttributeKvEntry cachedAttributeKvEntry = cachedAttributeValue.get();
-        } else {
+                return Optional.ofNullable(cachedAttributeKvEntry);
-            missCounter.increment();
+            } else {
-            return cacheExecutor.submit(() -> {
+                missCounter.increment();
                 var cacheTransaction = cache.newTransactionForKey(attributeCacheKey);
                 try {
                     Optional<AttributeKvEntry> result = attributesDao.find(tenantId, entityId, scope, attributeKey);
@@ -133,8 +133,8 @@ public class CachedAttributesService implements AttributesService {
                     log.debug("Could not find attribute from cache: [{}] [{}] [{}]", entityId, scope, attributeKey, e);
                     throw e;
                 }
-            });
+            }
-        }
+        });
     }
 
     @Override
@@ -207,7 +207,8 @@ public class CachedAttributesService implements AttributesService {
     @Override
     public ListenableFuture<List<AttributeKvEntry>> findAll(TenantId tenantId, EntityId entityId, String scope) {
         validate(entityId, scope);
-        return Futures.immediateFuture(attributesDao.findAll(tenantId, entityId, scope));
+        // We can`t watch on cache because the keys are unknown.
+        return jpaExecutorService.submit(() -> attributesDao.findAll(tenantId, entityId, scope));
     }
 
     @Override

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

@@ -27,6 +27,7 @@ import org.thingsboard.server.dao.exception.IncorrectParameterException;
 
 import java.util.List;
 import java.util.UUID;
+import java.util.function.Function;
 import java.util.regex.Pattern;
 
 public class Validator {
@@ -46,6 +47,19 @@ public class Validator {
         }
     }
 
+    /**
+     * This method validate <code>EntityId</code> entity id. If entity id is invalid than throw
+     * <code>IncorrectParameterException</code> exception
+     *
+     * @param entityId                  the entityId
+     * @param errorMessageFunction      the error message for exception that apply entityId
+     */
+    public static void validateEntityId(EntityId entityId, Function<EntityId, String> errorMessageFunction) {
+        if (entityId == null || entityId.getId() == null) {
+            throw new IncorrectParameterException(errorMessageFunction.apply(entityId));
+        }
+    }
+
     /**
      * This method validate <code>String</code> string. If string is invalid than throw
      * <code>IncorrectParameterException</code> exception
@@ -59,6 +73,18 @@ public class Validator {
         }
     }
 
+    /*
+     * This method validate <code>String</code> string. If string is invalid than throw
+     * <code>IncorrectParameterException</code> exception
+     *
+     * @param val                       the value
+     * @param errorMessageFunction      the error message function that apply value
+     */
+    public static void validateString(String val, Function<String, String> errorMessageFunction) {
+        if (val == null || val.isEmpty()) {
+            throw new IncorrectParameterException(errorMessageFunction.apply(val));
+        }
+    }
 
     /**
      * This method validate <code>long</code> value. If value isn't positive than throw

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

@@ -151,12 +151,12 @@ public class SqlTimeseriesLatestDao extends BaseAbstractSqlTimeseriesDao impleme
 
     @Override
     public ListenableFuture<Optional<TsKvEntry>> findLatestOpt(TenantId tenantId, EntityId entityId, String key) {
-        return Futures.immediateFuture(Optional.ofNullable(doFindLatest(entityId, key)));
+        return service.submit(() -> Optional.ofNullable(doFindLatest(entityId, key)));
     }
 
     @Override
     public ListenableFuture<TsKvEntry> findLatest(TenantId tenantId, EntityId entityId, String key) {
-        return Futures.immediateFuture(getLatestTsKvEntry(entityId, key));
+        return service.submit(() -> getLatestTsKvEntry(entityId, key));
     }
 
     @Override
@@ -221,36 +221,29 @@ public class SqlTimeseriesLatestDao extends BaseAbstractSqlTimeseriesDao impleme
     }
 
     protected ListenableFuture<TsKvLatestRemovingResult> getRemoveLatestFuture(TenantId tenantId, EntityId entityId, DeleteTsKvQuery query) {
-        TsKvEntry latest = doFindLatest(entityId, query.getKey());
+        ListenableFuture<TsKvEntry> latestFuture = service.submit(() -> doFindLatest(entityId, query.getKey()));
-
+        return Futures.transformAsync(latestFuture, latest -> {
-        if (latest == null) {
+            if (latest == null) {
-            return Futures.immediateFuture(new TsKvLatestRemovingResult(query.getKey(), false));
+                return Futures.immediateFuture(new TsKvLatestRemovingResult(query.getKey(), false));
-        }
+            }
-
+            boolean isRemoved = false;
-        long ts = latest.getTs();
+            long ts = latest.getTs();
-        ListenableFuture<Boolean> removedLatestFuture;
+            if (ts >= query.getStartTs() && ts < query.getEndTs()) {
-        if (ts >= query.getStartTs() && ts < query.getEndTs()) {
+                TsKvLatestEntity latestEntity = new TsKvLatestEntity();
-            TsKvLatestEntity latestEntity = new TsKvLatestEntity();
+                latestEntity.setEntityId(entityId.getId());
-            latestEntity.setEntityId(entityId.getId());
+                latestEntity.setKey(getOrSaveKeyId(query.getKey()));
-            latestEntity.setKey(getOrSaveKeyId(query.getKey()));
-            removedLatestFuture = service.submit(() -> {
                 tsKvLatestRepository.delete(latestEntity);
-                return true;
+                isRemoved = true;
-            });
+                if (query.getRewriteLatestIfDeleted()) {
-        } else {
+                    return getNewLatestEntryFuture(tenantId, entityId, query);
-            removedLatestFuture = Futures.immediateFuture(false);
+                }
-        }
-
-        return Futures.transformAsync(removedLatestFuture, isRemoved -> {
-            if (isRemoved && query.getRewriteLatestIfDeleted()) {
-                return getNewLatestEntryFuture(tenantId, entityId, query);
             }
             return Futures.immediateFuture(new TsKvLatestRemovingResult(query.getKey(), isRemoved));
         }, MoreExecutors.directExecutor());
     }
 
     protected ListenableFuture<List<TsKvEntry>> getFindAllLatestFuture(EntityId entityId) {
-        return Futures.immediateFuture(
+        return service.submit(() ->
                 DaoUtil.convertDataList(Lists.newArrayList(
                         searchTsKvLatestRepository.findAllByEntityId(entityId.getId()))));
     }

dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java

@@ -16,7 +16,6 @@
 package org.thingsboard.server.dao.timeseries;
 
 import com.google.common.base.Function;
-import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
@@ -43,6 +42,7 @@ import org.thingsboard.server.dao.entityview.EntityViewService;
 import org.thingsboard.server.dao.exception.IncorrectParameterException;
 import org.thingsboard.server.dao.service.Validator;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
@@ -126,18 +126,22 @@ public class BaseTimeseriesService implements TimeseriesService {
     @Override
     public ListenableFuture<List<TsKvEntry>> findLatest(TenantId tenantId, EntityId entityId, Collection<String> keys) {
         validate(entityId);
-        List<ListenableFuture<TsKvEntry>> futures = Lists.newArrayListWithExpectedSize(keys.size());
+        List<ListenableFuture<TsKvEntry>> futures = new ArrayList<>(keys.size());
-        keys.forEach(key -> Validator.validateString(key, "Incorrect key " + key));
+        keys.forEach(key -> Validator.validateString(key, k -> "Incorrect key " + k));
-        keys.forEach(key -> futures.add(timeseriesLatestDao.findLatest(tenantId, entityId, key)));
+        for (String key : keys) {
+            futures.add(timeseriesLatestDao.findLatest(tenantId, entityId, key));
+        }
         return Futures.allAsList(futures);
     }
 
     @Override
     public List<TsKvEntry> findLatestSync(TenantId tenantId, EntityId entityId, Collection<String> keys) {
         validate(entityId);
-        List<TsKvEntry> latestEntries = Lists.newArrayListWithExpectedSize(keys.size());
+        List<TsKvEntry> latestEntries = new ArrayList<>(keys.size());
-        keys.forEach(key -> Validator.validateString(key, "Incorrect key " + key));
+        keys.forEach(key -> Validator.validateString(key, k -> "Incorrect key " + k));
-        keys.forEach(key -> latestEntries.add(timeseriesLatestDao.findLatestSync(tenantId, entityId, key)));
+        for (String key : keys) {
+            latestEntries.add(timeseriesLatestDao.findLatestSync(tenantId, entityId, key));
+        }
         return latestEntries;
     }
 
@@ -165,7 +169,7 @@ public class BaseTimeseriesService implements TimeseriesService {
     @Override
     public ListenableFuture<Integer> save(TenantId tenantId, EntityId entityId, TsKvEntry tsKvEntry) {
         validate(entityId);
-        List<ListenableFuture<Integer>> futures = Lists.newArrayListWithExpectedSize(INSERTS_PER_ENTRY);
+        List<ListenableFuture<Integer>> futures = new ArrayList<>(INSERTS_PER_ENTRY);
         saveAndRegisterFutures(tenantId, futures, entityId, tsKvEntry, 0L);
         return Futures.transform(Futures.allAsList(futures), SUM_ALL_INTEGERS, MoreExecutors.directExecutor());
     }
@@ -182,7 +186,7 @@ public class BaseTimeseriesService implements TimeseriesService {
 
     private ListenableFuture<Integer> doSave(TenantId tenantId, EntityId entityId, List<TsKvEntry> tsKvEntries, long ttl, boolean saveLatest) {
         int inserts = saveLatest ? INSERTS_PER_ENTRY : INSERTS_PER_ENTRY_WITHOUT_LATEST;
-        List<ListenableFuture<Integer>> futures = Lists.newArrayListWithExpectedSize(tsKvEntries.size() * inserts);
+        List<ListenableFuture<Integer>> futures = new ArrayList<>(tsKvEntries.size() * inserts);
         for (TsKvEntry tsKvEntry : tsKvEntries) {
             if (saveLatest) {
                 saveAndRegisterFutures(tenantId, futures, entityId, tsKvEntry, ttl);
@@ -195,7 +199,7 @@ public class BaseTimeseriesService implements TimeseriesService {
 
     @Override
     public ListenableFuture<List<Void>> saveLatest(TenantId tenantId, EntityId entityId, List<TsKvEntry> tsKvEntries) {
-        List<ListenableFuture<Void>> futures = Lists.newArrayListWithExpectedSize(tsKvEntries.size());
+        List<ListenableFuture<Void>> futures = new ArrayList<>(tsKvEntries.size());
         for (TsKvEntry tsKvEntry : tsKvEntries) {
             futures.add(timeseriesLatestDao.saveLatest(tenantId, entityId, tsKvEntry));
         }
@@ -242,7 +246,7 @@ public class BaseTimeseriesService implements TimeseriesService {
     public ListenableFuture<List<TsKvLatestRemovingResult>> remove(TenantId tenantId, EntityId entityId, List<DeleteTsKvQuery> deleteTsKvQueries) {
         validate(entityId);
         deleteTsKvQueries.forEach(BaseTimeseriesService::validate);
-        List<ListenableFuture<TsKvLatestRemovingResult>> futures = Lists.newArrayListWithExpectedSize(deleteTsKvQueries.size() * DELETES_PER_ENTRY);
+        List<ListenableFuture<TsKvLatestRemovingResult>> futures = new ArrayList<>(deleteTsKvQueries.size() * DELETES_PER_ENTRY);
         for (DeleteTsKvQuery tsKvQuery : deleteTsKvQueries) {
             deleteAndRegisterFutures(tenantId, futures, entityId, tsKvQuery);
         }
@@ -252,7 +256,7 @@ public class BaseTimeseriesService implements TimeseriesService {
     @Override
     public ListenableFuture<List<TsKvLatestRemovingResult>> removeLatest(TenantId tenantId, EntityId entityId, Collection<String> keys) {
         validate(entityId);
-        List<ListenableFuture<TsKvLatestRemovingResult>> futures = Lists.newArrayListWithExpectedSize(keys.size());
+        List<ListenableFuture<TsKvLatestRemovingResult>> futures = new ArrayList<>(keys.size());
         for (String key : keys) {
             DeleteTsKvQuery query = new BaseDeleteTsKvQuery(key, 0, System.currentTimeMillis(), false);
             futures.add(timeseriesLatestDao.removeLatest(tenantId, entityId, query));
@@ -281,7 +285,7 @@ public class BaseTimeseriesService implements TimeseriesService {
     }
 
     private static void validate(EntityId entityId) {
-        Validator.validateEntityId(entityId, "Incorrect entityId " + entityId);
+        Validator.validateEntityId(entityId, id -> "Incorrect entityId " + id);
     }
 
     private void validate(ReadTsKvQuery query) {

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java

@@ -99,10 +99,10 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC
     private void safePutAttributes(TbContext ctx, TbMsg msg, ObjectNode msgDataNode, T entityId) {
         Set<TbPair<String, List<String>>> failuresPairSet = ConcurrentHashMap.newKeySet();
         var getKvEntryPairFutures = Futures.allAsList(
-                getLatestTelemetry(ctx, entityId, TbNodeUtils.processPatterns(config.getLatestTsKeyNames(), msg), failuresPairSet),
                 getAttrAsync(ctx, entityId, CLIENT_SCOPE, TbNodeUtils.processPatterns(config.getClientAttributeNames(), msg), failuresPairSet),
                 getAttrAsync(ctx, entityId, SHARED_SCOPE, TbNodeUtils.processPatterns(config.getSharedAttributeNames(), msg), failuresPairSet),
-                getAttrAsync(ctx, entityId, SERVER_SCOPE, TbNodeUtils.processPatterns(config.getServerAttributeNames(), msg), failuresPairSet)
+                getAttrAsync(ctx, entityId, SERVER_SCOPE, TbNodeUtils.processPatterns(config.getServerAttributeNames(), msg), failuresPairSet),
+                getLatestTelemetry(ctx, entityId, TbNodeUtils.processPatterns(config.getLatestTsKeyNames(), msg), failuresPairSet)
         );
         withCallback(getKvEntryPairFutures, futuresList -> {
             var msgMetaData = msg.getMetaData().copy();