From 59d0f36697c0e7bad424670b4b8c97a7ce3f3652 Mon Sep 17 00:00:00 2001
From: Sergey Matvienko <smatvienko@thingsboard.io>
Date: Fri, 8 Mar 2024 15:22:16 +0100
Subject: [PATCH] CachedAttributesService: replaced blocking calls and
 immediate futures with the true async

---
 .../attributes/CachedAttributesService.java   | 25 ++++++++++---------
 1 file changed, 13 insertions(+), 12 deletions(-)

diff --git a/dao/src/main/java/org/thingsboard/server/dao/attributes/CachedAttributesService.java b/dao/src/main/java/org/thingsboard/server/dao/attributes/CachedAttributesService.java
index 1666015a48..8058b0cee5 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/attributes/CachedAttributesService.java
+++ b/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);
-        TbCacheValueWrapper<AttributeKvEntry> cachedAttributeValue = cache.get(attributeCacheKey);
-        if (cachedAttributeValue != null) {
-            hitCounter.increment();
-            AttributeKvEntry cachedAttributeKvEntry = cachedAttributeValue.get();
-            return Futures.immediateFuture(Optional.ofNullable(cachedAttributeKvEntry));
-        } else {
-            missCounter.increment();
-            return cacheExecutor.submit(() -> {
+        return cacheExecutor.submit(() -> {
+            AttributeCacheKey attributeCacheKey = new AttributeCacheKey(scope, entityId, attributeKey);
+            TbCacheValueWrapper<AttributeKvEntry> cachedAttributeValue = cache.get(attributeCacheKey);
+            if (cachedAttributeValue != null) {
+                hitCounter.increment();
+                AttributeKvEntry cachedAttributeKvEntry = cachedAttributeValue.get();
+                return Optional.ofNullable(cachedAttributeKvEntry);
+            } else {
+                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