Merge pull request quarkusio#35977 from cescoffier/redis-cache-fault-…

…tolerance Recompute cached value when the Redis connection fails
gsmet · Sep 18, 2023 · f786f4a · f786f4a
2 parents 5a3a825 + c2b4860
commit f786f4a
Show file tree

Hide file tree

Showing 2 changed files with 77 additions and 14 deletions.
diff --git a/...ions/redis-cache/runtime/src/main/java/io/quarkus/cache/redis/runtime/RedisCacheImpl.java b/...ions/redis-cache/runtime/src/main/java/io/quarkus/cache/redis/runtime/RedisCacheImpl.java
@@ -1,5 +1,6 @@
 package io.quarkus.cache.redis.runtime;
 
+import java.net.ConnectException;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.List;
@@ -10,6 +11,8 @@
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 
+import org.jboss.logging.Logger;
+
 import io.quarkus.arc.Arc;
 import io.quarkus.arc.ArcContainer;
 import io.quarkus.cache.CacheException;
@@ -35,6 +38,8 @@
  */
 public class RedisCacheImpl extends AbstractCache implements RedisCache {
 
+    private static final Logger log = Logger.getLogger(RedisCacheImpl.class);
+
     private static final Map<String, Class<?>> PRIMITIVE_TO_CLASS_MAPPING = Map.of(
             "int", Integer.class,
             "byte", Byte.class,
@@ -123,6 +128,19 @@ private <K> String encodeKey(K key) {
         return new String(marshaller.encode(key), StandardCharsets.UTF_8);
     }
 
+    private <K, V> Uni<V> computeValue(K key, Function<K, V> valueLoader, boolean isWorkerThread) {
+        if (isWorkerThread) {
+            return Uni.createFrom().item(new Supplier<V>() {
+                @Override
+                public V get() {
+                    return valueLoader.apply(key);
+                }
+            }).runSubscriptionOn(MutinyHelper.blockingExecutor(vertx.getDelegate()));
+        } else {
+            return Uni.createFrom().item(valueLoader.apply(key));
+        }
+    }
+
     @Override
     public <K, V> Uni<V> get(K key, Class<V> clazz, Function<K, V> valueLoader) {
         // With optimistic locking:
@@ -148,17 +166,7 @@ public Uni<V> apply(V cached) throws Exception {
                                 if (cached != null) {
                                     return Uni.createFrom().item(new StaticSupplier<>(cached));
                                 } else {
-                                    Uni<V> uni;
-                                    if (isWorkerThread) {
-                                        uni = Uni.createFrom().item(new Supplier<V>() {
-                                            @Override
-                                            public V get() {
-                                                return valueLoader.apply(key);
-                                            }
-                                        }).runSubscriptionOn(MutinyHelper.blockingExecutor(vertx.getDelegate()));
-                                    } else {
-                                        uni = Uni.createFrom().item(valueLoader.apply(key));
-                                    }
+                                    Uni<V> uni = computeValue(key, valueLoader, isWorkerThread);
 
                                     return uni.onItem().call(new Function<V, Uni<?>>() {
                                         @Override
@@ -185,7 +193,15 @@ public Uni<?> apply(V value) {
                             }
                         }));
             }
-        });
+        })
+
+                .onFailure(ConnectException.class).recoverWithUni(new Function<Throwable, Uni<? extends V>>() {
+                    @Override
+                    public Uni<? extends V> apply(Throwable e) {
+                        log.warn("Unable to connect to Redis, recomputing cached value", e);
+                        return computeValue(key, valueLoader, isWorkerThread);
+                    }
+                });
     }
 
     @Override
@@ -215,7 +231,11 @@ public Uni<V> apply(RedisConnection connection) {
                             }
                         });
             }
-        });
+        })
+                .onFailure(ConnectException.class).recoverWithUni(e -> {
+                    log.warn("Unable to connect to Redis, recomputing cached value", e);
+                    return valueLoader.apply(key);
+                });
     }
 
     @Override

diff --git a/.../redis-cache/runtime/src/test/java/io/quarkus/cache/redis/runtime/RedisCacheImplTest.java b/.../redis-cache/runtime/src/test/java/io/quarkus/cache/redis/runtime/RedisCacheImplTest.java
@@ -29,7 +29,12 @@ class RedisCacheImplTest extends RedisCacheTestBase {
 
     @AfterEach
     void clear() {
-        redis.send(Request.cmd(Command.FLUSHALL).arg("SYNC")).await().atMost(Duration.ofSeconds(10));
+        try {
+            redis.send(Request.cmd(Command.FLUSHALL).arg("SYNC")).await()
+                    .atMost(Duration.ofSeconds(10));
+        } catch (Exception ignored) {
+            // ignored.
+        }
     }
 
     @Test
@@ -45,6 +50,18 @@ public void testPutInTheCache() {
         assertThat(r).isNotNull();
     }
 
+    @Test
+    public void testPutInTheCacheWithoutRedis() {
+        String k = UUID.randomUUID().toString();
+        RedisCacheInfo info = new RedisCacheInfo();
+        info.name = "foo";
+        info.valueType = String.class.getName();
+        info.expireAfterWrite = Optional.of(Duration.ofSeconds(2));
+        RedisCacheImpl cache = new RedisCacheImpl(info, vertx, redis, BLOCKING_ALLOWED);
+        server.close();
+        assertThat(cache.get(k, s -> "hello").await().indefinitely()).isEqualTo("hello");
+    }
+
     @Test
     public void testPutInTheCacheWithOptimisticLocking() {
         String k = UUID.randomUUID().toString();
@@ -355,6 +372,32 @@ void testAsyncGetWithDefaultType() {
                 }));
     }
 
+    @Test
+    void testAsyncGetWithDefaultTypeWithoutRedis() {
+        RedisCacheInfo info = new RedisCacheInfo();
+        info.name = "star-wars";
+        info.expireAfterWrite = Optional.of(Duration.ofSeconds(2));
+        info.valueType = Person.class.getName();
+        RedisCacheImpl cache = new RedisCacheImpl(info, vertx, redis, BLOCKING_ALLOWED);
+
+        server.close();
+
+        assertThat(cache
+                .getAsync("test",
+                        x -> Uni.createFrom().item(new Person("luke", "skywalker"))
+                                .runSubscriptionOn(Infrastructure.getDefaultExecutor()))
+                .await().indefinitely()).satisfies(p -> {
+                    assertThat(p.firstName).isEqualTo("luke");
+                    assertThat(p.lastName).isEqualTo("skywalker");
+                });
+
+        assertThat(cache.getAsync("test", x -> Uni.createFrom().item(new Person("leia", "organa")))
+                .await().indefinitely()).satisfies(p -> {
+                    assertThat(p.firstName).isEqualTo("leia");
+                    assertThat(p.lastName).isEqualTo("organa");
+                });
+    }
+
     @Test
     void testAsyncGetWithDefaultTypeWithOptimisticLocking() {
         RedisCacheInfo info = new RedisCacheInfo();