[automatic failover] Introduce fast failover mode

.gitignore

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

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/CommandObjects.java

@@ -74,6 +74,10 @@ public final CommandObject<String> ping() {
     return PING_COMMAND_OBJECT;
   }
 
+  public final CommandObject<String> echo(String msg) {
+    return new CommandObject<>(commandArguments(ECHO).add(msg), BuilderFactory.STRING);
+  }
+
   private final CommandObject<String> FLUSHALL_COMMAND_OBJECT = new CommandObject<>(commandArguments(FLUSHALL), BuilderFactory.STRING);
 
   public final CommandObject<String> flushAll() {

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

@@ -288,6 +288,10 @@ public void disconnect() {
     }
   }
 
+  public void forceDisconnect() throws IOException {
+    socket.close();
+  }
+
   public boolean isConnected() {
     return socket != null && socket.isBound() && !socket.isClosed() && socket.isConnected()
         && !socket.isInputShutdown() && !socket.isOutputShutdown();

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

@@ -7,6 +7,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.function.Supplier;
+import java.util.function.UnaryOperator;
 
 import redis.clients.jedis.annots.Experimental;
 import redis.clients.jedis.authentication.AuthXManager;
@@ -21,12 +22,15 @@
  */
 public class ConnectionFactory implements PooledObjectFactory<Connection> {
 
+  public interface MakerInjector extends UnaryOperator<Supplier<Connection>> {
+  };
+
   private static final Logger logger = LoggerFactory.getLogger(ConnectionFactory.class);
 
   private final JedisSocketFactory jedisSocketFactory;
   private final JedisClientConfig clientConfig;
   private final Cache clientSideCache;
-  private final Supplier<Connection> objectMaker;
+  private Supplier<Connection> objectMaker;
 
   private final AuthXEventListener authXEventListener;
 
@@ -73,6 +77,10 @@ private Supplier<Connection> connectionSupplier() {
         : () -> new CacheConnection(jedisSocketFactory, clientConfig, clientSideCache);
   }
 
+  public void injectMaker(MakerInjector injector) {
+    this.objectMaker = injector.apply(objectMaker);
+  }
+
   @Override
   public void activateObject(PooledObject<Connection> pooledConnection) throws Exception {
     // what to do ??

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

@@ -2,7 +2,9 @@
 
 import java.io.Serializable;
 
-public class HostAndPort implements Serializable {
+import redis.clients.jedis.mcf.Endpoint;
+
+public class HostAndPort implements Serializable, Endpoint {
 
     private static final long serialVersionUID = -519876229978427751L;
 
@@ -14,10 +16,12 @@ public HostAndPort(String host, int port) {
         this.port = port;
     }
 
+    @Override
     public String getHost() {
         return host;
     }
 
+    @Override
     public int getPort() {
         return port;
     }

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

@@ -12,6 +12,8 @@
 import redis.clients.jedis.exceptions.JedisConnectionException;
 import redis.clients.jedis.exceptions.JedisValidationException;
 import redis.clients.jedis.mcf.ConnectionFailoverException;
+import redis.clients.jedis.mcf.EchoStrategy;
+import redis.clients.jedis.mcf.HealthCheckStrategy;
 
 /**
  * @author Allen Terleto (aterleto)
@@ -31,6 +33,19 @@
 @Experimental
 public final class MultiClusterClientConfig {
 
+    /**
+     * Interface for creating HealthCheckStrategy instances for specific endpoints
+     */
+    public static interface StrategySupplier {
+        /**
+         * Creates a HealthCheckStrategy for the given endpoint.
+         * @param hostAndPort the endpoint to create a strategy for
+         * @param jedisClientConfig the client configuration, may be null for implementations that don't need it
+         * @return a HealthCheckStrategy instance
+         */
+        HealthCheckStrategy get(HostAndPort hostAndPort, JedisClientConfig jedisClientConfig);
+    }
+
     private static final int RETRY_MAX_ATTEMPTS_DEFAULT = 3;
     private static final int RETRY_WAIT_DURATION_DEFAULT = 500; // measured in milliseconds
     private static final int RETRY_WAIT_DURATION_EXPONENTIAL_BACKOFF_MULTIPLIER_DEFAULT = 2;
@@ -48,6 +63,9 @@ public final class MultiClusterClientConfig {
     private static final List<Class<? extends Throwable>> FALLBACK_EXCEPTIONS_DEFAULT = Arrays
         .asList(CallNotPermittedException.class, ConnectionFailoverException.class);
 
+    private static final long FAILBACK_CHECK_INTERVAL_DEFAULT = 5000; // 5 seconds
+    private static final long GRACE_PERIOD_DEFAULT = 10000; // 10 seconds
+
     private final ClusterConfig[] clusterConfigs;
 
     //////////// Retry Config - https://resilience4j.readme.io/docs/retry ////////////
@@ -129,7 +147,30 @@ public final class MultiClusterClientConfig {
 
     private List<Class<? extends Throwable>> fallbackExceptionList;
 
+    //////////// Failover Config ////////////
+
+    /** Whether to retry failed commands during failover */
+    private boolean retryOnFailover;
+
+    /** Whether failback is supported by client */
+    private boolean isFailbackSupported;
+
+    /** Interval in milliseconds to wait before attempting failback to a recovered cluster */
+    private long failbackCheckInterval;
+
+    /** 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");
+        for (ClusterConfig clusterConfig : clusterConfigs) {
+            if (clusterConfig == null)
+                throw new IllegalArgumentException("ClusterClientConfigs must not contain null elements");
+        }
         this.clusterConfigs = clusterConfigs;
     }
 
@@ -193,13 +234,40 @@ public List<Class<? extends Throwable>> getFallbackExceptionList() {
         return fallbackExceptionList;
     }
 
+    public boolean isRetryOnFailover() {
+        return retryOnFailover;
+    }
+
+    /** Whether failback is supported by client */
+    public boolean isFailbackSupported() {
+        return isFailbackSupported;
+    }
+
+    public long getFailbackCheckInterval() {
+        return failbackCheckInterval;
+    }
+
+    public long getGracePeriod() {
+        return gracePeriod;
+    }
+
+    public boolean isFastFailover() {
+        return fastFailover;
+    }
+
+    public static Builder builder(ClusterConfig[] clusterConfigs) {
+        return new Builder(clusterConfigs);
+    }
+
     public static class ClusterConfig {
 
-        private int priority;
         private HostAndPort hostAndPort;
         private JedisClientConfig clientConfig;
         private GenericObjectPoolConfig<Connection> connectionPoolConfig;
 
+        private float weight = 1.0f;
+        private StrategySupplier healthCheckStrategySupplier;
+
         public ClusterConfig(HostAndPort hostAndPort, JedisClientConfig clientConfig) {
             this.hostAndPort = hostAndPort;
             this.clientConfig = clientConfig;
@@ -212,25 +280,90 @@ public ClusterConfig(HostAndPort hostAndPort, JedisClientConfig clientConfig,
             this.connectionPoolConfig = connectionPoolConfig;
         }
 
-        public int getPriority() {
-            return priority;
-        }
-
-        private void setPriority(int priority) {
-            this.priority = priority;
+        private ClusterConfig(Builder builder) {
+            this.hostAndPort = builder.hostAndPort;
+            this.clientConfig = builder.clientConfig;
+            this.connectionPoolConfig = builder.connectionPoolConfig;
+            this.weight = builder.weight;
+            this.healthCheckStrategySupplier = builder.healthCheckStrategySupplier;
         }
 
         public HostAndPort getHostAndPort() {
             return hostAndPort;
         }
 
+        public static Builder builder(HostAndPort hostAndPort, JedisClientConfig clientConfig) {
+            return new Builder(hostAndPort, clientConfig);
+        }
+
         public JedisClientConfig getJedisClientConfig() {
             return clientConfig;
         }
 
         public GenericObjectPoolConfig<Connection> getConnectionPoolConfig() {
             return connectionPoolConfig;
         }
+
+        public float getWeight() {
+            return weight;
+        }
+
+        public StrategySupplier getHealthCheckStrategySupplier() {
+            return healthCheckStrategySupplier;
+        }
+
+        public static class Builder {
+            private HostAndPort hostAndPort;
+            private JedisClientConfig clientConfig;
+            private GenericObjectPoolConfig<Connection> connectionPoolConfig;
+
+            private float weight = 1.0f;
+            private StrategySupplier healthCheckStrategySupplier = EchoStrategy.DEFAULT;
+
+            public Builder(HostAndPort hostAndPort, JedisClientConfig clientConfig) {
+                this.hostAndPort = hostAndPort;
+                this.clientConfig = clientConfig;
+            }
+
+            public Builder connectionPoolConfig(GenericObjectPoolConfig<Connection> connectionPoolConfig) {
+                this.connectionPoolConfig = connectionPoolConfig;
+                return this;
+            }
+
+            public Builder weight(float weight) {
+                this.weight = weight;
+                return this;
+            }
+
+            public Builder healthCheckStrategySupplier(StrategySupplier healthCheckStrategySupplier) {
+                if (healthCheckStrategySupplier == null) {
+                    throw new IllegalArgumentException("healthCheckStrategySupplier must not be null");
+                }
+                this.healthCheckStrategySupplier = healthCheckStrategySupplier;
+                return this;
+            }
+
+            public Builder healthCheckStrategy(HealthCheckStrategy healthCheckStrategy) {
+                if (healthCheckStrategy == null) {
+                    throw new IllegalArgumentException("healthCheckStrategy must not be null");
+                }
+                this.healthCheckStrategySupplier = (hostAndPort, jedisClientConfig) -> healthCheckStrategy;
+                return this;
+            }
+
+            public Builder healthCheckEnabled(boolean healthCheckEnabled) {
+                if (!healthCheckEnabled) {
+                    this.healthCheckStrategySupplier = null;
+                } else if (healthCheckStrategySupplier == null) {
+                    this.healthCheckStrategySupplier = EchoStrategy.DEFAULT;
+                }
+                return this;
+            }
+
+            public ClusterConfig build() {
+                return new ClusterConfig(this);
+            }
+        }
     }
 
     public static class Builder {
@@ -253,14 +386,18 @@ public static class Builder {
         private List<Class> circuitBreakerIgnoreExceptionList = null;
         private List<Class<? extends Throwable>> fallbackExceptionList = FALLBACK_EXCEPTIONS_DEFAULT;
 
+        private boolean retryOnFailover = false;
+        private boolean isFailbackSupported = true;
+        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(
                 "ClusterClientConfigs are required for MultiClusterPooledConnectionProvider");
 
-            for (int i = 0; i < clusterConfigs.length; i++)
-                clusterConfigs[i].setPriority(i + 1);
-
             this.clusterConfigs = clusterConfigs;
         }
 
@@ -348,6 +485,31 @@ public Builder fallbackExceptionList(List<Class<? extends Throwable>> fallbackEx
             return this;
         }
 
+        public Builder retryOnFailover(boolean retryOnFailover) {
+            this.retryOnFailover = retryOnFailover;
+            return this;
+        }
+
+        public Builder failbackSupported(boolean supported) {
+            this.isFailbackSupported = supported;
+            return this;
+        }
+
+        public Builder failbackCheckInterval(long failbackCheckInterval) {
+            this.failbackCheckInterval = failbackCheckInterval;
+            return this;
+        }
+
+        public Builder gracePeriod(long gracePeriod) {
+            this.gracePeriod = gracePeriod;
+            return this;
+        }
+
+        public Builder fastFailover(boolean fastFailover) {
+            this.fastFailover = fastFailover;
+            return this;
+        }
+
         public MultiClusterClientConfig build() {
             MultiClusterClientConfig config = new MultiClusterClientConfig(this.clusterConfigs);
 
@@ -373,6 +535,12 @@ public MultiClusterClientConfig build() {
 
             config.fallbackExceptionList = this.fallbackExceptionList;
 
+            config.retryOnFailover = this.retryOnFailover;
+            config.isFailbackSupported = this.isFailbackSupported;
+            config.failbackCheckInterval = this.failbackCheckInterval;
+            config.gracePeriod = this.gracePeriod;
+            config.fastFailover = this.fastFailover;
+
             return config;
         }
     }