Json commands not exposed in AsyncCluster #3048 (#3049)

* Added missing ASYNC and REACTIVE interfaces for the cluster commands

* Handle container start failures more gracefully

* Attempt to start the containers three times before failing

Author: tishun

src/main/java/io/lettuce/core/cluster/api/async/RedisClusterAsyncCommands.java

@@ -37,11 +37,12 @@
  * @author dengliming
  * @since 4.0
  */
-public interface RedisClusterAsyncCommands<K, V> extends BaseRedisAsyncCommands<K, V>, RedisAclAsyncCommands<K, V>,
-        RedisFunctionAsyncCommands<K, V>, RedisGeoAsyncCommands<K, V>, RedisHashAsyncCommands<K, V>,
-        RedisHLLAsyncCommands<K, V>, RedisKeyAsyncCommands<K, V>, RedisListAsyncCommands<K, V>,
-        RedisScriptingAsyncCommands<K, V>, RedisServerAsyncCommands<K, V>, RedisSetAsyncCommands<K, V>,
-        RedisSortedSetAsyncCommands<K, V>, RedisStreamAsyncCommands<K, V>, RedisStringAsyncCommands<K, V> {
+public interface RedisClusterAsyncCommands<K, V>
+        extends BaseRedisAsyncCommands<K, V>, RedisAclAsyncCommands<K, V>, RedisFunctionAsyncCommands<K, V>,
+        RedisGeoAsyncCommands<K, V>, RedisHashAsyncCommands<K, V>, RedisHLLAsyncCommands<K, V>, RedisKeyAsyncCommands<K, V>,
+        RedisListAsyncCommands<K, V>, RedisScriptingAsyncCommands<K, V>, RedisServerAsyncCommands<K, V>,
+        RedisSetAsyncCommands<K, V>, RedisSortedSetAsyncCommands<K, V>, RedisStreamAsyncCommands<K, V>,
+        RedisStringAsyncCommands<K, V>, RedisJsonAsyncCommands<K, V> {
 
     /**
      * Set the default timeout for operations. A zero timeout value indicates to not time out.

src/main/java/io/lettuce/core/cluster/api/reactive/RedisClusterReactiveCommands.java

@@ -37,11 +37,12 @@
  * @author dengliming
  * @since 5.0
  */
-public interface RedisClusterReactiveCommands<K, V> extends BaseRedisReactiveCommands<K, V>, RedisAclReactiveCommands<K, V>,
-        RedisFunctionReactiveCommands<K, V>, RedisGeoReactiveCommands<K, V>, RedisHashReactiveCommands<K, V>,
-        RedisHLLReactiveCommands<K, V>, RedisKeyReactiveCommands<K, V>, RedisListReactiveCommands<K, V>,
-        RedisScriptingReactiveCommands<K, V>, RedisServerReactiveCommands<K, V>, RedisSetReactiveCommands<K, V>,
-        RedisSortedSetReactiveCommands<K, V>, RedisStreamReactiveCommands<K, V>, RedisStringReactiveCommands<K, V> {
+public interface RedisClusterReactiveCommands<K, V>
+        extends BaseRedisReactiveCommands<K, V>, RedisAclReactiveCommands<K, V>, RedisFunctionReactiveCommands<K, V>,
+        RedisGeoReactiveCommands<K, V>, RedisHashReactiveCommands<K, V>, RedisHLLReactiveCommands<K, V>,
+        RedisKeyReactiveCommands<K, V>, RedisListReactiveCommands<K, V>, RedisScriptingReactiveCommands<K, V>,
+        RedisServerReactiveCommands<K, V>, RedisSetReactiveCommands<K, V>, RedisSortedSetReactiveCommands<K, V>,
+        RedisStreamReactiveCommands<K, V>, RedisStringReactiveCommands<K, V>, RedisJsonReactiveCommands<K, V> {
 
     /**
      * Set the default timeout for operations. A zero timeout value indicates to not time out.

src/test/java/io/lettuce/core/RedisContainerIntegrationTests.java

@@ -16,7 +16,6 @@
 import org.testcontainers.junit.jupiter.Testcontainers;
 
 import java.io.File;
-import java.io.IOException;
 
 @Testcontainers
 public class RedisContainerIntegrationTests {
@@ -27,22 +26,41 @@ public class RedisContainerIntegrationTests {
 
     private static final String REDIS_STACK_CLUSTER = "clustered-stack";
 
+    private static Exception initializationException;
+
     public static ComposeContainer CLUSTERED_STACK = new ComposeContainer(
             new File("src/test/resources/docker/docker-compose.yml")).withExposedService(REDIS_STACK_CLUSTER, 36379)
                     .withExposedService(REDIS_STACK_CLUSTER, 36380).withExposedService(REDIS_STACK_CLUSTER, 36381)
                     .withExposedService(REDIS_STACK_CLUSTER, 36382).withExposedService(REDIS_STACK_CLUSTER, 36383)
                     .withExposedService(REDIS_STACK_CLUSTER, 36384).withExposedService(REDIS_STACK_STANDALONE, 6379)
                     .withLocalCompose(true);
 
-    @BeforeAll
-    public static void setup() throws IOException, InterruptedException {
+    // Singleton container pattern - start the containers only once
+    // See https://java.testcontainers.org/test_framework_integration/manual_lifecycle_control/#singleton-containers
+    static {
+        int attempts = 0;
+
         // In case you need to debug the container uncomment these lines to redirect the output
         CLUSTERED_STACK.withLogConsumer(REDIS_STACK_CLUSTER, (OutputFrame frame) -> LOGGER.debug(frame.getUtf8String()));
         CLUSTERED_STACK.withLogConsumer(REDIS_STACK_STANDALONE, (OutputFrame frame) -> LOGGER.debug(frame.getUtf8String()));
 
         CLUSTERED_STACK.waitingFor(REDIS_STACK_CLUSTER,
                 Wait.forLogMessage(".*Background RDB transfer terminated with success.*", 1));
-        CLUSTERED_STACK.start();
+        do {
+            try {
+                CLUSTERED_STACK.start();
+            } catch (Exception e) {
+                initializationException = e;
+            }
+            // Attempt to stabilize the pipeline - sometime the `docker compose up` fails randomly
+        } while (initializationException != null && attempts++ < 3);
+    }
+
+    @BeforeAll
+    public static void checkContainerInitialization() {
+        if (initializationException != null) {
+            throw new IllegalStateException("Failed to initialize containers", initializationException);
+        }
     }
 
 }

src/test/java/io/lettuce/core/json/RedisJsonClusterIntegrationTests.java

@@ -10,6 +10,8 @@
 import io.lettuce.core.RedisContainerIntegrationTests;
 import io.lettuce.core.RedisURI;
 import io.lettuce.core.cluster.RedisClusterClient;
+import io.lettuce.core.cluster.api.async.RedisClusterAsyncCommands;
+import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands;
 import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;
 import io.lettuce.core.json.arguments.JsonGetArgs;
 import io.lettuce.core.json.arguments.JsonMsetArgs;
@@ -21,6 +23,7 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
+import reactor.test.StepVerifier;
 
 import java.io.IOException;
 import java.nio.file.Files;
@@ -30,6 +33,7 @@
 import java.time.format.DateTimeFormatter;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.ExecutionException;
 
 import static io.lettuce.TestTags.INTEGRATION_TEST;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -114,6 +118,22 @@ void jsonArrinsert(String path) {
         assertThat(arrayIndex.get(0).longValue()).isEqualTo(3L);
     }
 
+    @Test
+    void jsonArrLenAsyncAndReactive() throws ExecutionException, InterruptedException {
+        RedisClusterAsyncCommands<String, String> asyncCommands = client.connect().async();
+        RedisClusterReactiveCommands<String, String> reactiveCommands = client.connect().reactive();
+
+        JsonPath myPath = JsonPath.of(MOUNTAIN_BIKES_V1);
+
+        List<Long> poppedJson = asyncCommands.jsonArrlen(BIKES_INVENTORY, myPath).get();
+        assertThat(poppedJson).hasSize(1);
+        assertThat(poppedJson.get(0).longValue()).isEqualTo(3);
+
+        StepVerifier.create(reactiveCommands.jsonArrlen(BIKES_INVENTORY, myPath)).consumeNextWith(actual -> {
+            assertThat(actual).isEqualTo(3);
+        }).verifyComplete();
+    }
+
     @ParameterizedTest(name = "With {0} as path")
     @ValueSource(strings = { MOUNTAIN_BIKES_V1, MOUNTAIN_BIKES_V2 })
     void jsonArrlen(String path) {

src/test/java/io/lettuce/core/json/RedisJsonIntegrationTests.java

@@ -14,6 +14,7 @@
 import io.lettuce.core.RedisURI;
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.api.async.RedisAsyncCommands;
+import io.lettuce.core.api.reactive.RedisReactiveCommands;
 import io.lettuce.core.api.sync.RedisCommands;
 import io.lettuce.core.codec.ByteArrayCodec;
 import io.lettuce.core.codec.StringCodec;
@@ -26,6 +27,7 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -130,6 +132,22 @@ void jsonArrlen(String path) {
         assertThat(poppedJson.get(0).longValue()).isEqualTo(3);
     }
 
+    @Test
+    void jsonArrLenAsyncAndReactive() throws ExecutionException, InterruptedException {
+        RedisAsyncCommands<String, String> asyncCommands = client.connect().async();
+        RedisReactiveCommands<String, String> reactiveCommands = client.connect().reactive();
+
+        JsonPath myPath = JsonPath.of(MOUNTAIN_BIKES_V1);
+
+        List<Long> poppedJson = asyncCommands.jsonArrlen(BIKES_INVENTORY, myPath).get();
+        assertThat(poppedJson).hasSize(1);
+        assertThat(poppedJson.get(0).longValue()).isEqualTo(3);
+
+        StepVerifier.create(reactiveCommands.jsonArrlen(BIKES_INVENTORY, myPath)).consumeNextWith(actual -> {
+            assertThat(actual).isEqualTo(3);
+        }).verifyComplete();
+    }
+
     @ParameterizedTest(name = "With {0} as path")
     @ValueSource(strings = { MOUNTAIN_BIKES_V1, MOUNTAIN_BIKES_V2 })
     void jsonArrpop(String path) {