- fix failing scenario test

- downgrade logback version for slf4j compatibility
- increase timeouts for faultInjector

Author: atakavci

pom.xml

@@ -143,7 +143,7 @@
 		<dependency>
 			<groupId>ch.qos.logback</groupId>
 			<artifactId>logback-classic</artifactId>
-			<version>1.3.15</version>
+			<version>1.2.12</version>
 			<scope>test</scope>
 		</dependency>
 		<dependency>

src/main/java/redis/clients/jedis/providers/MultiClusterPooledConnectionProvider.java

@@ -650,6 +650,7 @@ public static class Cluster {
 
         // Grace period tracking
         private volatile long gracePeriodEndsAt = 0;
+        private final Logger log = LoggerFactory.getLogger(getClass());
 
         public Cluster(ConnectionPool connectionPool, Retry retry, CircuitBreaker circuitBreaker, float weight,
             MultiClusterClientConfig multiClusterClientConfig) {
@@ -693,6 +694,7 @@ public float getWeight() {
 
         public boolean isCBForcedOpen() {
             if (circuitBreaker.getState() == State.FORCED_OPEN && !isInGracePeriod()) {
+                log.info("Transitioning circuit breaker from FORCED_OPEN to CLOSED state due to end of grace period!");
                 circuitBreaker.transitionToClosedState();
             }
             return circuitBreaker.getState() == CircuitBreaker.State.FORCED_OPEN;

src/test/java/redis/clients/jedis/scenario/ActiveActiveFailoverTest.java

@@ -8,11 +8,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import redis.clients.jedis.*;
+import redis.clients.jedis.MultiClusterClientConfig.ClusterConfig;
 import redis.clients.jedis.providers.MultiClusterPooledConnectionProvider;
 import redis.clients.jedis.exceptions.JedisConnectionException;
 
 import java.io.IOException;
-import java.time.Duration;
 import java.time.Instant;
 import java.util.HashMap;
 import java.util.Map;
@@ -25,10 +25,7 @@
 import static org.junit.jupiter.api.Assertions.fail;
 import static org.junit.jupiter.api.Assumptions.assumeTrue;
 
-@Tags({
-    @Tag("failover"),
-    @Tag("scenario")
-})
+@Tags({ @Tag("failover"), @Tag("scenario") })
 public class ActiveActiveFailoverTest {
   private static final Logger log = LoggerFactory.getLogger(ActiveActiveFailoverTest.class);
 
@@ -52,13 +49,13 @@ public void testFailover() {
     MultiClusterClientConfig.ClusterConfig[] clusterConfig = new MultiClusterClientConfig.ClusterConfig[2];
 
     JedisClientConfig config = endpoint.getClientConfigBuilder()
-        .socketTimeoutMillis(RecommendedSettings.DEFAULT_TIMEOUT_MS)
-        .connectionTimeoutMillis(RecommendedSettings.DEFAULT_TIMEOUT_MS).build();
+      .socketTimeoutMillis(RecommendedSettings.DEFAULT_TIMEOUT_MS)
+      .connectionTimeoutMillis(RecommendedSettings.DEFAULT_TIMEOUT_MS).build();
 
-    clusterConfig[0] = new MultiClusterClientConfig.ClusterConfig(endpoint.getHostAndPort(0),
-        config, RecommendedSettings.poolConfig);
-    clusterConfig[1] = new MultiClusterClientConfig.ClusterConfig(endpoint.getHostAndPort(1),
-        config, RecommendedSettings.poolConfig);
+    clusterConfig[0] = ClusterConfig.builder(endpoint.getHostAndPort(0), config)
+      .connectionPoolConfig(RecommendedSettings.poolConfig).weight(1.0f).build();
+    clusterConfig[1] = ClusterConfig.builder(endpoint.getHostAndPort(1), config)
+      .connectionPoolConfig(RecommendedSettings.poolConfig).weight(0.5f).build();
 
     MultiClusterClientConfig.Builder builder = new MultiClusterClientConfig.Builder(clusterConfig);
 
@@ -67,6 +64,10 @@ public void testFailover() {
     builder.circuitBreakerSlidingWindowMinCalls(1);
     builder.circuitBreakerFailureRateThreshold(10.0f); // percentage of failures to trigger circuit breaker
 
+    builder.failbackSupported(true);
+    builder.failbackCheckInterval(1000);
+    builder.gracePeriod(10000);
+
     builder.retryWaitDuration(10);
     builder.retryMaxAttempts(1);
     builder.retryWaitDurationExponentialBackoffMultiplier(1);
@@ -79,24 +80,30 @@ class FailoverReporter implements Consumer<String> {
 
       Instant failoverAt = null;
 
+      boolean failbackHappened = false;
+
+      Instant failbackAt = null;
+
       public String getCurrentClusterName() {
         return currentClusterName;
       }
 
       @Override
       public void accept(String clusterName) {
         this.currentClusterName = clusterName;
-        log.info(
-            "\n\n====FailoverEvent=== \nJedis failover to cluster: {}\n====FailoverEvent===\n\n",
-            clusterName);
-
-        failoverHappened = true;
-        failoverAt = Instant.now();
+        log.info("\n\n====FailoverEvent=== \nJedis failover to cluster: {}\n====FailoverEvent===\n\n", clusterName);
+
+        if (failoverHappened) {
+          failbackHappened = true;
+          failbackAt = Instant.now();
+        } else {
+          failoverHappened = true;
+          failoverAt = Instant.now();
+        }
       }
     }
 
-    MultiClusterPooledConnectionProvider provider = new MultiClusterPooledConnectionProvider(
-        builder.build());
+    MultiClusterPooledConnectionProvider provider = new MultiClusterPooledConnectionProvider(builder.build());
     FailoverReporter reporter = new FailoverReporter();
     provider.setClusterFailoverPostProcessor(reporter);
     provider.setActiveCluster(endpoint.getHostAndPort(0));
@@ -117,15 +124,17 @@ public void accept(String clusterName) {
       int retryingDelay = 5;
       while (true) {
         try {
-          Map<String, String> executionInfo = new HashMap<String, String>() {{
-            put("threadId", String.valueOf(threadId));
-            put("cluster", reporter.getCurrentClusterName());
-          }};
+          Map<String, String> executionInfo = new HashMap<String, String>() {
+            {
+              put("threadId", String.valueOf(threadId));
+              put("cluster", reporter.getCurrentClusterName());
+            }
+          };
           client.xadd("execution_log", StreamEntryID.NEW_ENTRY, executionInfo);
 
           if (attempt > 0) {
             log.info("Thread {} recovered after {} ms. Threads still not recovered: {}", threadId,
-                attempt * retryingDelay, retryingThreadsCounter.decrementAndGet());
+              attempt * retryingDelay, retryingThreadsCounter.decrementAndGet());
           }
 
           break;
@@ -134,15 +143,13 @@ public void accept(String clusterName) {
           if (reporter.failoverHappened) {
             long failedCommands = failedCommandsAfterFailover.incrementAndGet();
             lastFailedCommandAt.set(Instant.now());
-            log.warn(
-                "Thread {} failed to execute command after failover. Failed commands after failover: {}",
-                threadId, failedCommands);
+            log.warn("Thread {} failed to execute command after failover. Failed commands after failover: {}", threadId,
+              failedCommands);
           }
 
           if (attempt == 0) {
             long failedThreads = retryingThreadsCounter.incrementAndGet();
-            log.warn("Thread {} failed to execute command. Failed threads: {}", threadId,
-                failedThreads);
+            log.warn("Thread {} failed to execute command. Failed threads: {}", threadId, failedThreads);
           }
           try {
             Thread.sleep(retryingDelay);
@@ -153,20 +160,21 @@ public void accept(String clusterName) {
         }
       }
       return true;
-    }, 18);
+    }, 4);
     fakeApp.setKeepExecutingForSeconds(30);
     Thread t = new Thread(fakeApp);
     t.start();
 
     HashMap<String, Object> params = new HashMap<>();
     params.put("bdb_id", endpoint.getBdbId());
-    params.put("rlutil_command", "pause_bdb");
+    params.put("actions",
+      "[{\"type\":\"execute_rlutil_command\",\"params\":{\"rlutil_command\":\"pause_bdb\"}},{\"type\":\"wait\",\"params\":{\"wait_time\":\"15\"}},{\"type\":\"execute_rlutil_command\",\"params\":{\"rlutil_command\":\"resume_bdb\"}}]");
 
     FaultInjectionClient.TriggerActionResponse actionResponse = null;
 
     try {
-      log.info("Triggering bdb_pause");
-      actionResponse = faultClient.triggerAction("execute_rlutil_command", params);
+      log.info("Triggering bdb_pause + wait 15 seconds + bdb_resume");
+      actionResponse = faultClient.triggerAction("sequence_of_actions", params);
     } catch (IOException e) {
       fail("Fault Injection Server error:" + e.getMessage());
     }
@@ -182,15 +190,17 @@ public void accept(String clusterName) {
 
     ConnectionPool pool = provider.getCluster(endpoint.getHostAndPort(0)).getConnectionPool();
 
-    log.info("First connection pool state: active: {}, idle: {}", pool.getNumActive(),
-        pool.getNumIdle());
-    log.info("Full failover time: {} s",
-        Duration.between(reporter.failoverAt, lastFailedCommandAt.get()).getSeconds());
+    log.info("First connection pool state: active: {}, idle: {}", pool.getNumActive(), pool.getNumIdle());
+    log.info("Failover happened at: {}", reporter.failoverAt);
+    log.info("Failback happened at: {}", reporter.failbackAt);
+    log.info("Last failed command at: {}", lastFailedCommandAt.get());
 
     assertEquals(0, pool.getNumActive());
     assertTrue(fakeApp.capturedExceptions().isEmpty());
+    assertTrue(reporter.failoverHappened);
+    assertTrue(reporter.failbackHappened);
 
     client.close();
   }
 
-}
+}

src/test/java/redis/clients/jedis/scenario/FaultInjectionClient.java

@@ -89,8 +89,8 @@ public boolean isCompleted(Duration checkInterval, Duration delayAfter, Duration
 
   private static CloseableHttpClient getHttpClient() {
     RequestConfig requestConfig = RequestConfig.custom()
-        .setConnectionRequestTimeout(5000, TimeUnit.MILLISECONDS)
-        .setResponseTimeout(5000, TimeUnit.MILLISECONDS).build();
+        .setConnectionRequestTimeout(10000, TimeUnit.MILLISECONDS)
+        .setResponseTimeout(10000, TimeUnit.MILLISECONDS).build();
 
     return HttpClientBuilder.create()
         .setDefaultRequestConfig(requestConfig).build();