Keep track of pending Redis get calls in client side caching to avoid race with invalidations. #3481

src/main/java/io/lettuce/core/support/caching/CacheAccessor.java

@@ -8,7 +8,7 @@
  *
  * @param <K> Key type.
  * @param <V> Value type.
- * @author Mark Paluch
+ * @author Mark Paluch, Ivar Henckel
  * @since 6.0
  */
 public interface CacheAccessor<K, V> {
@@ -22,7 +22,7 @@ public interface CacheAccessor<K, V> {
      * @return a {@link CacheAccessor} backed by a {@link Map} implementation.
      */
     static <K, V> CacheAccessor<K, V> forMap(Map<K, V> map) {
-        return new MapCacheAccessor<>(map);
+        return new PendingAwareCacheAccessor<>(map);
     }
 
     /**
@@ -59,4 +59,16 @@ static <K, V> CacheAccessor<K, V> forMap(Map<K, V> map) {
      */
     void evict(K key);
 
+    /**
+     * Mark a key as pending.
+     * <p>
+     * By marking keys as pending, i.e. Redis get operation in progress, it's possible to block cache updates that would
+     * otherwise overwrite simultaneous invalidates.
+     *
+     * @param key the key to be marked as pending.
+     */
+    default void setPending(K key) {
+        // Do nothing.
+    }
+
 }

src/main/java/io/lettuce/core/support/caching/ClientSideCaching.java

@@ -124,6 +124,7 @@ public V get(K key) {
         V value = cacheAccessor.get(key);
 
         if (value == null) {
+            cacheAccessor.setPending(key);
             value = redisCache.get(key);
 
             if (value != null) {
@@ -140,6 +141,7 @@ public V get(K key, Callable<V> valueLoader) {
         V value = cacheAccessor.get(key);
 
         if (value == null) {
+            cacheAccessor.setPending(key);
             value = redisCache.get(key);
 
             if (value == null) {

src/main/java/io/lettuce/core/support/caching/PendingAwareCacheAccessor.java

@@ -0,0 +1,47 @@
+package io.lettuce.core.support.caching;
+
+import java.util.Map;
+
+/**
+ * {@link CacheAccessor} implementation for {@link Map}-based cache implementations.
+ *
+ * @param <K> Key type.
+ * @param <V> Value type.
+ * @author Mark Paluch, Ivar Henckel
+ * @since 6.0
+ */
+class PendingAwareCacheAccessor<K, V> implements CacheAccessor<K, V> {
+
+    private static final Object REDIS_IN_PROGRESS = new Object();
+
+    private final Map<K, V> map;
+
+    PendingAwareCacheAccessor(Map<K, V> map) {
+        this.map = map;
+    }
+
+    @Override
+    public V get(K key) {
+        V value = map.get(key);
+        if (value == REDIS_IN_PROGRESS) {
+            return null;
+        }
+        return value;
+    }
+
+    @Override
+    public void put(K key, V value) {
+        map.replace(key, (V) REDIS_IN_PROGRESS, value);
+    }
+
+    @Override
+    public void evict(K key) {
+        map.remove(key);
+    }
+
+    @Override
+    public void setPending(K key) {
+        map.put(key, (V) REDIS_IN_PROGRESS);
+    }
+
+}