Add test for Redis timeouts

service/src/test/java/org/whispersystems/textsecuregcm/redis/FaultTolerantRedisClusterTest.java

@@ -7,6 +7,7 @@ package org.whispersystems.textsecuregcm.redis;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
@@ -22,8 +23,11 @@ import io.lettuce.core.cluster.pubsub.StatefulRedisClusterPubSubConnection;
 import io.lettuce.core.event.EventBus;
 import io.lettuce.core.resource.ClientResources;
 import java.time.Duration;
+import java.util.concurrent.ExecutionException;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
 import org.whispersystems.textsecuregcm.configuration.CircuitBreakerConfiguration;
 import org.whispersystems.textsecuregcm.configuration.RetryConfiguration;
 import reactor.core.publisher.Flux;
@@ -38,7 +42,8 @@ class FaultTolerantRedisClusterTest {
   public void setUp() {
     final RedisClusterClient clusterClient = mock(RedisClusterClient.class);
     final StatefulRedisClusterConnection<String, String> clusterConnection = mock(StatefulRedisClusterConnection.class);
-        final StatefulRedisClusterPubSubConnection<String, String> pubSubConnection  = mock(StatefulRedisClusterPubSubConnection.class);
+    final StatefulRedisClusterPubSubConnection<String, String> pubSubConnection = mock(
+        StatefulRedisClusterPubSubConnection.class);
     final ClientResources clientResources = mock(ClientResources.class);
     final EventBus eventBus = mock(EventBus.class);
 
@@ -97,6 +102,43 @@ class FaultTolerantRedisClusterTest {
         .thenThrow(new RedisCommandTimeoutException())
         .thenReturn("value");
 
-        assertThrows(RedisCommandTimeoutException.class, () -> faultTolerantCluster.withCluster(connection -> connection.sync().get("key")));
+    assertThrows(RedisCommandTimeoutException.class,
+        () -> faultTolerantCluster.withCluster(connection -> connection.sync().get("key")));
+
   }
+
+  @Nested
+  class WithRealCluster {
+
+    private static final Duration TIMEOUT = Duration.ofMillis(50);
+
+    private static final RetryConfiguration retryConfiguration = new RetryConfiguration();
+
+    static {
+      retryConfiguration.setMaxAttempts(1);
+      retryConfiguration.setWaitDuration(50);
+    }
+
+    @RegisterExtension
+    static final RedisClusterExtension REDIS_CLUSTER_EXTENSION = RedisClusterExtension.builder()
+        .retryConfiguration(retryConfiguration)
+        .timeout(TIMEOUT)
+        .build();
+
+    @Test
+    void testTimeout() {
+      final FaultTolerantRedisCluster cluster = REDIS_CLUSTER_EXTENSION.getRedisCluster();
+
+      assertTimeoutPreemptively(Duration.ofSeconds(1), () -> {
+        final ExecutionException asyncException = assertThrows(ExecutionException.class,
+            () -> cluster.withCluster(connection -> connection.async().blpop(TIMEOUT.toMillis() * 2, "key")).get());
+        assertTrue(asyncException.getCause() instanceof RedisCommandTimeoutException);
+
+        assertThrows(RedisCommandTimeoutException.class,
+            () -> cluster.withCluster(connection -> connection.sync().blpop(TIMEOUT.toMillis() * 2, "key")));
+      });
+
+    }
+  }
+
 }

service/src/test/java/org/whispersystems/textsecuregcm/redis/RedisClusterExtension.java

@@ -32,14 +32,23 @@ import org.whispersystems.textsecuregcm.util.RedisClusterUtil;
 import redis.embedded.RedisServer;
 import redis.embedded.exceptions.EmbeddedRedisException;
 
-public class RedisClusterExtension implements BeforeAllCallback, BeforeEachCallback, AfterAllCallback, AfterEachCallback {
+public class RedisClusterExtension implements BeforeAllCallback, BeforeEachCallback, AfterAllCallback,
+    AfterEachCallback {
 
+  private static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(2);
   private static final int NODE_COUNT = 2;
 
   private static final RedisServer[] CLUSTER_NODES = new RedisServer[NODE_COUNT];
 
+  private final Duration timeout;
+  private final RetryConfiguration retryConfiguration;
   private FaultTolerantRedisCluster redisCluster;
 
+  public RedisClusterExtension(final Duration timeout, final RetryConfiguration retryConfiguration) {
+    this.timeout = timeout;
+    this.retryConfiguration = retryConfiguration;
+  }
+
 
   public static RedisClusterExtensionBuilder builder() {
     return new RedisClusterExtensionBuilder();
@@ -78,9 +87,9 @@ public class RedisClusterExtension implements BeforeAllCallback, BeforeEachCallb
 
     redisCluster = new FaultTolerantRedisCluster("test-cluster",
         RedisClusterClient.create(urls.stream().map(RedisURI::create).collect(Collectors.toList())),
-        Duration.ofSeconds(2),
+        timeout,
         new CircuitBreakerConfiguration(),
-        new RetryConfiguration());
+        retryConfiguration);
 
     redisCluster.useCluster(connection -> {
       boolean setAll = false;
@@ -208,11 +217,24 @@ public class RedisClusterExtension implements BeforeAllCallback, BeforeEachCallb
 
   public static class RedisClusterExtensionBuilder {
 
+    private Duration timeout = DEFAULT_TIMEOUT;
+    private RetryConfiguration retryConfiguration = new RetryConfiguration();
+
     private RedisClusterExtensionBuilder() {
     }
 
+    RedisClusterExtensionBuilder timeout(Duration timeout) {
+      this.timeout = timeout;
+      return this;
+    }
+
+    RedisClusterExtensionBuilder retryConfiguration(RetryConfiguration retryConfiguration) {
+      this.retryConfiguration = retryConfiguration;
+      return this;
+    }
+
     public RedisClusterExtension build() {
-      return new RedisClusterExtension();
+      return new RedisClusterExtension(timeout, retryConfiguration);
     }
   }
 }