Redis cache: make blocking executions unordered

When the Redis cache is invoked from an ordered Vert.x blocking execution,
which happens for example with SmallRye GraphQL or with chained caching
(one blocking `@CacheResult` method invoking other blocking `@CacheResult`
method), the Redis cache ends up hanging. This is because the next execution
cannot start until the previous execution finishes, but the previous execution
waits for the next execution.

This commit fixes that by making the Redis cache blocking executions unordered.

(cherry picked from commit 1fd2a9a)

extensions/redis-cache/deployment/src/test/java/io/quarkus/cache/redis/deployment/ChainedRedisCacheTest.java

@@ -0,0 +1,49 @@
+package io.quarkus.cache.redis.deployment;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.List;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.arc.Arc;
+import io.quarkus.cache.CacheResult;
+import io.quarkus.redis.datasource.RedisDataSource;
+import io.quarkus.test.QuarkusUnitTest;
+
+public class ChainedRedisCacheTest {
+    @RegisterExtension
+    static final QuarkusUnitTest TEST = new QuarkusUnitTest()
+            .withApplicationRoot(jar -> jar.addClasses(ChainedCachedService.class, TestUtil.class));
+
+    @Inject
+    ChainedCachedService chainedCachedService;
+
+    @Test
+    public void test() {
+        RedisDataSource redisDataSource = Arc.container().select(RedisDataSource.class).get();
+        List<String> allKeysAtStart = TestUtil.allRedisKeys(redisDataSource);
+
+        assertEquals("fubar:42", chainedCachedService.cache1("fubar"));
+
+        List<String> allKeysAtEnd = TestUtil.allRedisKeys(redisDataSource);
+        assertEquals(allKeysAtStart.size() + 2, allKeysAtEnd.size());
+    }
+
+    @ApplicationScoped
+    public static class ChainedCachedService {
+        @CacheResult(cacheName = "cache1")
+        public String cache1(String key) {
+            return key + ":" + cache2(42);
+        }
+
+        @CacheResult(cacheName = "cache2")
+        public int cache2(int value) {
+            return value;
+        }
+    }
+}

extensions/redis-cache/runtime/src/main/java/io/quarkus/cache/redis/runtime/RedisCacheImpl.java

@@ -141,7 +141,7 @@ private <K, V> Uni<V> computeValue(K key, Function<K, V> valueLoader, boolean is
                 public V get() {
                     return valueLoader.apply(key);
                 }
-            }).runSubscriptionOn(MutinyHelper.blockingExecutor(vertx.getDelegate()));
+            }).runSubscriptionOn(MutinyHelper.blockingExecutor(vertx.getDelegate(), false));
         } else {
             return Uni.createFrom().item(valueLoader.apply(key));
         }
@@ -205,8 +205,8 @@ public Uni<?> apply(V value) {
                                                 result = set(connection, encodedKey, encodedValue).replaceWith(value);
                                             }
                                             if (isWorkerThread) {
-                                                return result
-                                                        .runSubscriptionOn(MutinyHelper.blockingExecutor(vertx.getDelegate()));
+                                                return result.runSubscriptionOn(
+                                                        MutinyHelper.blockingExecutor(vertx.getDelegate(), false));
                                             }
                                             return result;
                                         }