cache: Transaction aware cache to synchronize cache put/evict operations with on…

…going Spring-managed transactions.

@@ -77,13 +77,13 @@
             <artifactId>logback-classic</artifactId>
             <artifactId>logback-classic</artifactId>
         </dependency>
         </dependency>
         <dependency>
         <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
             <scope>test</scope>
             <scope>test</scope>
         </dependency>
         </dependency>
         <dependency>
         <dependency>
-            <groupId>org.mockito</groupId>
-            <artifactId>mockito-core</artifactId>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
             <scope>test</scope>
             <scope>test</scope>
         </dependency>
         </dependency>
     </dependencies>
     </dependencies>

@@ -27,6 +27,7 @@ import org.springframework.cache.CacheManager;
 import org.springframework.cache.annotation.EnableCaching;
 import org.springframework.cache.annotation.EnableCaching;
 import org.springframework.cache.caffeine.CaffeineCache;
 import org.springframework.cache.caffeine.CaffeineCache;
 import org.springframework.cache.support.SimpleCacheManager;
 import org.springframework.cache.support.SimpleCacheManager;
+import org.springframework.cache.transaction.TransactionAwareCacheManagerProxy;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Configuration;
 
 
@@ -47,9 +48,14 @@ public class CaffeineCacheConfiguration {
 
 
     private Map<String, CacheSpecs> specs;
     private Map<String, CacheSpecs> specs;
 
 
+
+    /**
+     * Transaction aware CaffeineCache implementation with TransactionAwareCacheManagerProxy
+     * to synchronize cache put/evict operations with ongoing Spring-managed transactions.
+     */
     @Bean
     @Bean
     public CacheManager cacheManager() {
     public CacheManager cacheManager() {
-        log.trace("Initializing cache: {}", Arrays.toString(RemovalCause.values()));
+        log.trace("Initializing cache: {} specs {}", Arrays.toString(RemovalCause.values()), specs);
         SimpleCacheManager manager = new SimpleCacheManager();
         SimpleCacheManager manager = new SimpleCacheManager();
         if (specs != null) {
         if (specs != null) {
             List<CaffeineCache> caches =
             List<CaffeineCache> caches =
@@ -59,7 +65,11 @@ public class CaffeineCacheConfiguration {
                             .collect(Collectors.toList());
                             .collect(Collectors.toList());
             manager.setCaches(caches);
             manager.setCaches(caches);
         }
         }
-        return manager;
+
+        //SimpleCacheManager is not a bean (will be wrapped), so call initializeCaches manually
+        manager.initializeCaches();
+
+        return new TransactionAwareCacheManagerProxy(manager);
     }
     }
 
 
     private CaffeineCache buildCache(String name, CacheSpecs cacheSpec) {
     private CaffeineCache buildCache(String name, CacheSpecs cacheSpec) {

@@ -79,13 +79,19 @@ public abstract class TBRedisCacheConfiguration {
 
 
     protected abstract JedisConnectionFactory loadFactory();
     protected abstract JedisConnectionFactory loadFactory();
 
 
+    /**
+     * Transaction aware RedisCacheManager.
+     * Enable RedisCaches to synchronize cache put/evict operations with ongoing Spring-managed transactions.
+     */
     @Bean
     @Bean
     public CacheManager cacheManager(RedisConnectionFactory cf) {
     public CacheManager cacheManager(RedisConnectionFactory cf) {
         DefaultFormattingConversionService redisConversionService = new DefaultFormattingConversionService();
         DefaultFormattingConversionService redisConversionService = new DefaultFormattingConversionService();
         RedisCacheConfiguration.registerDefaultConverters(redisConversionService);
         RedisCacheConfiguration.registerDefaultConverters(redisConversionService);
         registerDefaultConverters(redisConversionService);
         registerDefaultConverters(redisConversionService);
         RedisCacheConfiguration configuration = RedisCacheConfiguration.defaultCacheConfig().withConversionService(redisConversionService);
         RedisCacheConfiguration configuration = RedisCacheConfiguration.defaultCacheConfig().withConversionService(redisConversionService);
-        return RedisCacheManager.builder(cf).cacheDefaults(configuration).build();
+        return RedisCacheManager.builder(cf).cacheDefaults(configuration)
+                .transactionAware()
+                .build();
     }
     }
 
 
     @Bean
     @Bean

+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.cache;
+
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cache.CacheManager;
+import org.springframework.cache.transaction.TransactionAwareCacheDecorator;
+import org.springframework.cache.transaction.TransactionAwareCacheManagerProxy;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@ExtendWith(SpringExtension.class)
+@ContextConfiguration(classes = CaffeineCacheConfiguration.class)
+@EnableConfigurationProperties
+@TestPropertySource(properties = {
+        "cache.type=caffeine",
+        "caffeine.specs.relations.timeToLiveInMinutes=1440",
+        "caffeine.specs.relations.maxSize=0",
+        "caffeine.specs.devices.timeToLiveInMinutes=60",
+        "caffeine.specs.devices.maxSize=100"})
+@Slf4j
+public class CaffeineCacheConfigurationTest {
+
+    @Autowired
+    CacheManager cacheManager;
+
+    @Test
+    public void verifyTransactionAwareCacheManagerProxy() {
+        assertThat(cacheManager).isInstanceOf(TransactionAwareCacheManagerProxy.class);
+    }
+
+    @Test
+    public void givenCacheConfig_whenCacheManagerReady_thenVerifyExistedCachesWithTransactionAwareCacheDecorator() {
+        assertThat(cacheManager.getCache("relations")).isInstanceOf(TransactionAwareCacheDecorator.class);
+        assertThat(cacheManager.getCache("devices")).isInstanceOf(TransactionAwareCacheDecorator.class);
+    }
+
+    @Test
+    public void givenCacheConfig_whenCacheManagerReady_thenVerifyNonExistedCaches() {
+        assertThat(cacheManager.getCache("rainbows_and_unicorns")).isNull();
+    }
+}

+<?xml version="1.0" encoding="UTF-8" ?>
+
+<configuration>
+    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{ISO8601} [%thread] %-5level %logger{36} - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <logger name="org.thingsboard.server.cache" level="TRACE"/>
+
+    <root level="INFO">
+        <appender-ref ref="console"/>
+    </root>
+</configuration>