- introduce clusterSwitchEvent and drop clusterFailover post processor

- fix broken echostrategy due to connection issue
- make healtthCheckStrategy closable and close on
- adding fastfailover mode to config and provider
- add local failover tests for total failover duration

Author: atakavci

.gitignore

@@ -17,3 +17,4 @@ tags
 redis-git
 appendonlydir/
 .DS_Store
+.vscode/settings.json

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

@@ -161,6 +161,9 @@ public static interface StrategySupplier {
     /** Grace period in milliseconds to keep clusters disabled after they become unhealthy */
     private long gracePeriod;
 
+    /** Whether to force terminate connections forcefully on failover */
+    private boolean fastFailover;
+
     public MultiClusterClientConfig(ClusterConfig[] clusterConfigs) {
         if (clusterConfigs == null || clusterConfigs.length < 1) throw new JedisValidationException(
             "ClusterClientConfigs are required for MultiClusterPooledConnectionProvider");
@@ -248,6 +251,10 @@ public long getGracePeriod() {
         return gracePeriod;
     }
 
+    public boolean isFastFailover() {
+        return fastFailover;
+    }
+
     public static Builder builder(ClusterConfig[] clusterConfigs) {
         return new Builder(clusterConfigs);
     }
@@ -384,6 +391,8 @@ public static class Builder {
         private long failbackCheckInterval = FAILBACK_CHECK_INTERVAL_DEFAULT;
         private long gracePeriod = GRACE_PERIOD_DEFAULT;
 
+        private boolean fastFailover = false;
+
         public Builder(ClusterConfig[] clusterConfigs) {
 
             if (clusterConfigs == null || clusterConfigs.length < 1) throw new JedisValidationException(
@@ -496,6 +505,11 @@ public Builder gracePeriod(long gracePeriod) {
             return this;
         }
 
+        public Builder fastFailover(boolean fastFailover) {
+            this.fastFailover = fastFailover;
+            return this;
+        }
+
         public MultiClusterClientConfig build() {
             MultiClusterClientConfig config = new MultiClusterClientConfig(this.clusterConfigs);
 
@@ -525,6 +539,7 @@ public MultiClusterClientConfig build() {
             config.isFailbackSupported = this.isFailbackSupported;
             config.failbackCheckInterval = this.failbackCheckInterval;
             config.gracePeriod = this.gracePeriod;
+            config.fastFailover = this.fastFailover;
 
             return config;
         }

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

@@ -59,12 +59,7 @@ protected void clusterFailover(CircuitBreaker circuitBreaker) {
                 // Iterating the active cluster will allow subsequent calls to the executeCommand() to use the next
                 // cluster's connection pool - according to the configuration's prioritization/order/weight
                 // int activeMultiClusterIndex = provider.incrementActiveMultiClusterIndex1();
-                if (provider.iterateActiveCluster() != null) {
-
-                    // Implementation is optionally provided during configuration. Typically, used for
-                    // activeMultiClusterIndex persistence or custom logging
-                    provider.runClusterFailoverPostProcessor(provider.getCluster());
-                }
+                provider.iterateActiveCluster(SwitchReason.CIRCUIT_BREAKER);
             }
             // this check relies on the fact that many failover attempts can hit with the same CB,
             // only the first one will trigger a failover, and make the CB FORCED_OPEN.

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

@@ -0,0 +1,29 @@
+package redis.clients.jedis.mcf;
+
+import redis.clients.jedis.providers.MultiClusterPooledConnectionProvider.Cluster;
+
+public class ClusterSwitchEventArgs {
+
+    private final SwitchReason reason;
+    private final String ClusterName;
+    private final Endpoint Endpoint;
+
+    public ClusterSwitchEventArgs(SwitchReason reason, Endpoint endpoint, Cluster cluster) {
+        this.reason = reason;
+        this.ClusterName = cluster.getCircuitBreaker().getName();
+        this.Endpoint = endpoint;
+    }
+
+    public SwitchReason getReason() {
+        return reason;
+    }
+
+    public String getClusterName() {
+        return ClusterName;
+    }
+
+    public Endpoint getEndpoint() {
+        return Endpoint;
+    }
+
+}

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

@@ -1,10 +1,8 @@
 package redis.clients.jedis.mcf;
 
-import redis.clients.jedis.ConnectionFactory;
 import redis.clients.jedis.HostAndPort;
 import redis.clients.jedis.JedisClientConfig;
 import redis.clients.jedis.UnifiedJedis;
-import redis.clients.jedis.exceptions.JedisConnectionException;
 import redis.clients.jedis.MultiClusterClientConfig.StrategySupplier;
 
 public class EchoStrategy implements HealthCheckStrategy {
@@ -20,12 +18,7 @@ public EchoStrategy(HostAndPort hostAndPort, JedisClientConfig jedisClientConfig
     public EchoStrategy(HostAndPort hostAndPort, JedisClientConfig jedisClientConfig, int interval, int timeout) {
         this.interval = interval;
         this.timeout = timeout;
-        ConnectionFactory connFactory = new ConnectionFactory(hostAndPort, jedisClientConfig);
-        try {
-            this.jedis = new UnifiedJedis(connFactory.makeObject().getObject());
-        } catch (Exception e) {
-            throw new JedisConnectionException("HealthCheck connection Failed!", e);
-        }
+        this.jedis = new UnifiedJedis(hostAndPort, jedisClientConfig);
     }
 
     @Override
@@ -43,6 +36,11 @@ public HealthStatus doHealthCheck(Endpoint endpoint) {
         return "HealthCheck".equals(jedis.echo("HealthCheck")) ? HealthStatus.HEALTHY : HealthStatus.UNHEALTHY;
     }
 
+    @Override
+    public void close() {
+        jedis.close();
+    }
+
     public static final StrategySupplier DEFAULT = (hostAndPort, jedisClientConfig) -> {
         return new EchoStrategy(hostAndPort, jedisClientConfig);
     };

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

@@ -12,8 +12,13 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 public class HealthCheck {
 
+    private static final Logger log = LoggerFactory.getLogger(HealthCheck.class);
+
     private Endpoint endpoint;
     private HealthCheckStrategy strategy;
     private AtomicReference<SimpleEntry<Long, HealthStatus>> statusRef = new AtomicReference<SimpleEntry<Long, HealthStatus>>();
@@ -43,6 +48,7 @@ public void start() {
     }
 
     public void stop() {
+        strategy.close();
         this.statusChangeCallback = null;
         scheduler.shutdown();
         executor.shutdown();
@@ -67,6 +73,7 @@ private void healthCheck() {
         Future<?> future = executor.submit(() -> {
             HealthStatus newStatus = strategy.doHealthCheck(endpoint);
             safeUpdate(me, newStatus);
+            log.trace("Health check completed for {} with status {}", endpoint, newStatus);
         });
 
         try {
@@ -75,11 +82,13 @@ private void healthCheck() {
             // Cancel immediately on timeout or exec exception
             future.cancel(true);
             safeUpdate(me, HealthStatus.UNHEALTHY);
+            log.warn("Health check timed out or failed for {}", endpoint, e);
         } catch (InterruptedException e) {
             // Health check thread was interrupted
             future.cancel(true);
             safeUpdate(me, HealthStatus.UNHEALTHY);
             Thread.currentThread().interrupt(); // Restore interrupted status
+            log.warn("Health check interrupted for {}", endpoint, e);
         }
     }
 

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

@@ -1,11 +1,16 @@
 package redis.clients.jedis.mcf;
 
-public interface HealthCheckStrategy {
+import java.io.Closeable;
+
+public interface HealthCheckStrategy extends Closeable {
 
     int getInterval();
 
     int getTimeout();
 
     HealthStatus doHealthCheck(Endpoint endpoint);
 
+    default void close() {
+    }
+
 }

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

@@ -0,0 +1,5 @@
+package redis.clients.jedis.mcf;
+
+public enum SwitchReason {
+    HEALTH_CHECK, CIRCUIT_BREAKER, FAILBACK, FORCED
+}