[automatic-failover] Implement failover retry for in-flight commands (#4175)

* Implement failover retry for in-flight commands

This change adds support for retrying in-flight commands when a Redis connection
fails and the client automatically fails over to another cluster. The feature
is configurable through the FailoverOptions builder.

Key changes:
- Added retryFailedInflightCommands option to FailoverOptions
- Implemented retry logic in CircuitBreakerCommandExecutor
- Added integration tests to verify both retry and no-retry behavior
- Created utility methods for test setup and configuration

This enhancement improves resilience for long-running commands like blocking
operations, allowing them to transparently continue on the failover cluster
without client-side errors.

* Update src/main/java/redis/clients/jedis/mcf/CircuitBreakerCommandExecutor.java

Co-authored-by: Copilot <[email protected]>
# Conflicts:
#	src/main/java/redis/clients/jedis/mcf/CircuitBreakerCommandExecutor.java

* format

format & clean-up

* fix FailoverIntegrationTest.testInflightCommandsAreNotRetriedAfterFailover

    blpop timeout should be less than async command timeout to prevent completing with java.util.concurrent.TimeoutException instead of actuall command failure

* Address comments from review
   - rename retryFailedInflightCommands->retryOnFailover
   - remove check for CB in OPEN state

* remove unused method

---------

Co-authored-by: Copilot <[email protected]>

Author: ggivo

src/main/java/redis/clients/jedis/MultiClusterClientConfig.java

@@ -11,7 +11,7 @@
 import redis.clients.jedis.annots.Experimental;
 import redis.clients.jedis.exceptions.JedisConnectionException;
 import redis.clients.jedis.exceptions.JedisValidationException;
-
+import redis.clients.jedis.mcf.ConnectionFailoverException;
 
 /**
  * @author Allen Terleto (aterleto)
@@ -43,7 +43,7 @@ public final class MultiClusterClientConfig {
     private static final float CIRCUIT_BREAKER_SLOW_CALL_RATE_THRESHOLD_DEFAULT = 100.0f; // measured as percentage
     private static final List<Class> CIRCUIT_BREAKER_INCLUDED_EXCEPTIONS_DEFAULT = Arrays.asList(JedisConnectionException.class);
 
-    private static final List<Class<? extends Throwable>> FALLBACK_EXCEPTIONS_DEFAULT = Arrays.asList(CallNotPermittedException.class);
+    private static final List<Class<? extends Throwable>> FALLBACK_EXCEPTIONS_DEFAULT = Arrays.asList(CallNotPermittedException.class, ConnectionFailoverException.class);
 
     private final ClusterConfig[] clusterConfigs;
 

src/main/java/redis/clients/jedis/UnifiedJedis.java

@@ -30,6 +30,7 @@
 import redis.clients.jedis.json.Path2;
 import redis.clients.jedis.json.JsonObjectMapper;
 import redis.clients.jedis.mcf.CircuitBreakerCommandExecutor;
+import redis.clients.jedis.mcf.FailoverOptions;
 import redis.clients.jedis.mcf.MultiClusterPipeline;
 import redis.clients.jedis.mcf.MultiClusterTransaction;
 import redis.clients.jedis.params.*;
@@ -237,7 +238,19 @@ public UnifiedJedis(ConnectionProvider provider, int maxAttempts, Duration maxTo
    */
   @Experimental
   public UnifiedJedis(MultiClusterPooledConnectionProvider provider) {
-    this(new CircuitBreakerCommandExecutor(provider), provider);
+    this(new CircuitBreakerCommandExecutor(provider, FailoverOptions.builder().build()), provider);
+  }
+
+  /**
+   * Constructor which supports multiple cluster/database endpoints each with their own isolated connection pool.
+   * <p>
+   * With this Constructor users can seamlessly failover to Disaster Recovery (DR), Backup, and Active-Active cluster(s)
+   * by using simple configuration which is passed through from Resilience4j - https://resilience4j.readme.io/docs
+   * <p>
+   */
+  @Experimental
+  public UnifiedJedis(MultiClusterPooledConnectionProvider provider, FailoverOptions failoverOptions) {
+    this(new CircuitBreakerCommandExecutor(provider, failoverOptions), provider);
   }
 
   /**

src/main/java/redis/clients/jedis/mcf/CircuitBreakerCommandExecutor.java

@@ -24,8 +24,12 @@
 public class CircuitBreakerCommandExecutor extends CircuitBreakerFailoverBase
     implements CommandExecutor {
 
-  public CircuitBreakerCommandExecutor(MultiClusterPooledConnectionProvider provider) {
+  private final FailoverOptions options;
+
+  public CircuitBreakerCommandExecutor(MultiClusterPooledConnectionProvider provider,
+      FailoverOptions options) {
     super(provider);
+    this.options = options != null ? options : FailoverOptions.builder().build();
   }
 
   @Override
@@ -49,9 +53,31 @@ public <T> T executeCommand(CommandObject<T> commandObject) {
   private <T> T handleExecuteCommand(CommandObject<T> commandObject, Cluster cluster) {
     try (Connection connection = cluster.getConnection()) {
       return connection.executeCommand(commandObject);
+    } catch (Exception e) {
+
+      if (retryOnFailover() && !isActiveCluster(cluster)
+          && isCircuitBreakerTrackedException(e, cluster.getCircuitBreaker())) {
+        throw new ConnectionFailoverException(
+            "Command failed during failover: " + cluster.getCircuitBreaker().getName(), e);
+      }
+
+      throw e;
     }
   }
 
+  private boolean isCircuitBreakerTrackedException(Exception e, CircuitBreaker cb) {
+    return cb.getCircuitBreakerConfig().getRecordExceptionPredicate().test(e);
+  }
+
+  private boolean retryOnFailover() {
+    return options.isRetryOnFailover();
+  }
+
+  private boolean isActiveCluster(Cluster cluster) {
+    Cluster activeCluster = provider.getCluster();
+    return activeCluster != null && activeCluster.equals(cluster);
+  }
+
   /**
    * Functional interface wrapped in retry and circuit breaker logic to handle open circuit breaker
    * failure scenarios

src/main/java/redis/clients/jedis/mcf/ConnectionFailoverException.java

@@ -0,0 +1,9 @@
+package redis.clients.jedis.mcf;
+
+import redis.clients.jedis.exceptions.JedisException;
+
+public class ConnectionFailoverException extends JedisException {
+  public ConnectionFailoverException(String s, Exception e) {
+    super(s, e);
+  }
+}

src/main/java/redis/clients/jedis/mcf/FailoverOptions.java

@@ -0,0 +1,59 @@
+package redis.clients.jedis.mcf;
+
+import redis.clients.jedis.annots.Experimental;
+
+/**
+ * Configuration options for CircuitBreakerCommandExecutor
+ */
+@Experimental
+public class FailoverOptions {
+  private final boolean retryOnFailover;
+
+  private FailoverOptions(Builder builder) {
+    this.retryOnFailover = builder.retryOnFailover;
+  }
+
+  /**
+   * Gets whether to retry failed commands during failover
+   * @return true if retry is enabled, false otherwise
+   */
+  public boolean isRetryOnFailover() {
+    return retryOnFailover;
+  }
+
+  /**
+   * Creates a new builder with default options
+   * @return a new builder
+   */
+  public static Builder builder() {
+    return new Builder();
+  }
+
+  /**
+   * Builder for FailoverOptions
+   */
+  public static class Builder {
+    private boolean retryOnFailover = false;
+
+    private Builder() {
+    }
+
+    /**
+     * Sets whether to retry failed commands during failover
+     * @param retry true to retry, false otherwise
+     * @return this builder for method chaining
+     */
+    public Builder retryOnFailover(boolean retry) {
+      this.retryOnFailover = retry;
+      return this;
+    }
+
+    /**
+     * Builds a new FailoverOptions instance with the configured options
+     * @return a new FailoverOptions instance
+     */
+    public FailoverOptions build() {
+      return new FailoverOptions(this);
+    }
+  }
+}