[automatic failover] Introduce fast failover mode - a thread-sync-free approach with builders (#4226)

* - weighted cluster seleciton
- Healtstatus manager with initial listener and registration logic
- pluggable health checker strategy  introduced,  these are draft NoOpStrategy, EchoStrategy, LagAwareStrategy,
- fix failing tests impacted from weighted clusters

* - add builder for ClusterConfig
- add echo ot CommandObjects and UnifiedJEdis
- improve StrategySupplier by accepting jedisclientconfig
- adapt EchoStrategy to StrategySupplier. Now it handles the creation of connection by accepting endpoint and JedisClientConfig
- make healthchecks disabled by default
- drop noOpStrategy
-  add unit&integration tests for health check

* - fix naming

* clean up and mark override methods

* fix link in javadoc

* fix formatting

* - fix double registered listeners in healtstatusmgr
- clear redundant catch
- replace failover options and drop failoveroptions class
- remove forced_unhealthy from healthstatus
- fix failback check
- add disabled flag to cluster
- update/fix related tests

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

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

* - add remove endpoints

* - replace cluster disabled with failbackCandidate
- replace failback enabled with failbacksupported in client
- fix formatting
- set defaults

* - remove failback candidate
- fix failing tests

* - fix remove logic
- fix failing tests

* - periodic failback checks
- introduce graceperiod
- fix issue when CB is forced_open and gracePeriod is completed

* - introduce StatusTracker with purpose of waiting initial healthcheck results during consturction of provider
- add HealthStatus.UNKNOWN as default for Cluster
- handle status changes in order of events during initialization
- add tests for status tracker and orderingof events
- fix impacted unit&integ tests

* - introduce forceActiveCluster by duration
- fix formatting

* - fix failing tests by waiting on clusters to get healthy

* - fix failing scenario test
- downgrade logback version for slf4j compatibility
- increase timeouts for faultInjector

* - adressing reviews and feedback

* - fix formatting

* - fix formatting

* - get rid of the queue and event ordering for healthstatus change in MultiClusterPooledConnectionProvider
- add test for init and post init events
- fix failing tests

* - replace use of reflection with helper methods
- fix failing tests due to method name change

* - 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

* - introduce fastfailover using  objectMaker injection into connectionFactory

* - polish

* - cleanup

* - improve healtcheck thread visibility

* - introduce TrackingConnectionPool with FailFastConnectionFactory
- added builders to connection and connectionFactory
- introduce initializtionTracker to track list of connections during their construction.

* - return broken source as usual
- do not throw exception is failover already happening

* - unblock waiting threads

* - failover by closing the pool

* - formatting

* - check waiters and active/idle connections to force disconnect

* - add builder to trackingconnectionpool

* - fix failing tests due to mocked ctor for trackingConnectionPool

* - replace initTracker with split initializtion of conn

* - refactor on builders and ctors

* - undo format

* - clena up

* - add exceptions to logs

* - add max wait duration and minConsecutiveSuccessCount to healthCheckStrategy

* - replace  'sleep' with 'await', feedback from @a-TODO-rov

* - fix status tracker tests

---------

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

Author: atakavci

.gitignore

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

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

@@ -33,6 +33,39 @@
 
 public class Connection implements Closeable {
 
+  public static class Builder {
+    private JedisSocketFactory socketFactory;
+    private JedisClientConfig clientConfig;
+
+    public Builder socketFactory(JedisSocketFactory socketFactory) {
+      this.socketFactory = socketFactory;
+      return this;
+    }
+
+    public Builder clientConfig(JedisClientConfig clientConfig) {
+      this.clientConfig = clientConfig;
+      return this;
+    }
+
+    public JedisSocketFactory getSocketFactory() {
+      return socketFactory;
+    }
+
+    public JedisClientConfig getClientConfig() {
+      return clientConfig;
+    }
+
+    public Connection build() {
+      Connection conn = new Connection(this);
+      conn.initializeFromClientConfig();
+      return conn;
+    }
+  }
+
+  public static Builder builder() {
+    return new Builder();
+  }
+
   private ConnectionPool memberOf;
   protected RedisProtocol protocol;
   private final JedisSocketFactory socketFactory;
@@ -48,6 +81,7 @@ public class Connection implements Closeable {
   protected String version;
   private AtomicReference<RedisCredentials> currentCredentials = new AtomicReference<>(null);
   private AuthXManager authXManager;
+  private JedisClientConfig clientConfig;
 
   public Connection() {
     this(Protocol.DEFAULT_HOST, Protocol.DEFAULT_PORT);
@@ -67,16 +101,19 @@ public Connection(final HostAndPort hostAndPort, final JedisClientConfig clientC
 
   public Connection(final JedisSocketFactory socketFactory) {
     this.socketFactory = socketFactory;
-    this.authXManager = null;
   }
 
   public Connection(final JedisSocketFactory socketFactory, JedisClientConfig clientConfig) {
     this.socketFactory = socketFactory;
-    this.soTimeout = clientConfig.getSocketTimeoutMillis();
-    this.infiniteSoTimeout = clientConfig.getBlockingSocketTimeoutMillis();
+    this.clientConfig = clientConfig;
     initializeFromClientConfig(clientConfig);
   }
 
+  protected Connection(Builder builder) {
+    this.socketFactory = builder.getSocketFactory();
+    this.clientConfig = builder.getClientConfig();
+  }
+
   @Override
   public String toString() {
     return getClass().getSimpleName() + "{" + socketFactory + "}";
@@ -288,6 +325,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();
@@ -450,8 +491,15 @@ private static boolean validateClientInfo(String info) {
     return true;
   }
 
+  public void initializeFromClientConfig() {
+    this.initializeFromClientConfig(clientConfig);
+  }
+
   protected void initializeFromClientConfig(final JedisClientConfig config) {
     try {
+      this.soTimeout = config.getSocketTimeoutMillis();
+      this.infiniteSoTimeout = config.getBlockingSocketTimeoutMillis();
+
       connect();
 
       protocol = config.getRedisProtocol();

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

@@ -21,56 +21,137 @@
  */
 public class ConnectionFactory implements PooledObjectFactory<Connection> {
 
+  public static class Builder {
+    private JedisClientConfig clientConfig;
+    private Connection.Builder connectionBuilder;
+    private JedisSocketFactory jedisSocketFactory;
+    private Cache cache;
+    private HostAndPort hostAndPort;
+
+    // Fluent API methods (preferred)
+    public Builder clientConfig(JedisClientConfig clientConfig) {
+      this.clientConfig = clientConfig;
+      return this;
+    }
+
+    public Builder connectionBuilder(Connection.Builder connectionBuilder) {
+      this.connectionBuilder = connectionBuilder;
+      return this;
+    }
+
+    public Builder socketFactory(JedisSocketFactory jedisSocketFactory) {
+      this.jedisSocketFactory = jedisSocketFactory;
+      return this;
+    }
+
+    public Builder cache(Cache cache) {
+      this.cache = cache;
+      return this;
+    }
+
+    public Builder hostAndPort(HostAndPort hostAndPort) {
+      this.hostAndPort = hostAndPort;
+      return this;
+    }
+
+    public Connection.Builder getConnectionBuilder() {
+      return connectionBuilder;
+    }
+
+    public JedisSocketFactory getJedisSocketFactory() {
+      return jedisSocketFactory;
+    }
+
+    public JedisClientConfig getClientConfig() {
+      return clientConfig;
+    }
+
+    public Cache getCache() {
+      return cache;
+    }
+
+    public ConnectionFactory build() {
+      withDefaults();
+      return new ConnectionFactory(this);
+    }
+
+    private Builder withDefaults() {
+      if (jedisSocketFactory == null) {
+        this.jedisSocketFactory = createDefaultSocketFactory();
+      }
+      if (connectionBuilder == null) {
+        this.connectionBuilder = createDefaultConnectionBuilder();
+      }
+      return this;
+    }
+
+    private JedisSocketFactory createDefaultSocketFactory() {
+      if (clientConfig == null) {
+        clientConfig = DefaultJedisClientConfig.builder().build();
+      }
+      if (hostAndPort == null) {
+        throw new IllegalStateException("HostAndPort is required when no socketFactory is provided");
+      }
+      return new DefaultJedisSocketFactory(hostAndPort, clientConfig);
+    }
+
+    private Connection.Builder createDefaultConnectionBuilder() {
+      Connection.Builder connBuilder = cache == null ? Connection.builder() : CacheConnection.builder(cache);
+      connBuilder.socketFactory(jedisSocketFactory).clientConfig(clientConfig);
+      return connBuilder;
+    }
+  }
+
+  public static Builder builder() {
+    return new Builder();
+  }
+
   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 Connection.Builder connectionBuilder;
 
-  private final AuthXEventListener authXEventListener;
+  private AuthXEventListener authXEventListener;
 
   public ConnectionFactory(final HostAndPort hostAndPort) {
-    this(hostAndPort, DefaultJedisClientConfig.builder().build(), null);
+    this(builder().hostAndPort(hostAndPort).withDefaults());
   }
 
   public ConnectionFactory(final HostAndPort hostAndPort, final JedisClientConfig clientConfig) {
-    this(hostAndPort, clientConfig, null);
+    this(builder().hostAndPort(hostAndPort).clientConfig(clientConfig).withDefaults());
   }
 
   @Experimental
-  public ConnectionFactory(final HostAndPort hostAndPort, final JedisClientConfig clientConfig,
-      Cache csCache) {
-    this(new DefaultJedisSocketFactory(hostAndPort, clientConfig), clientConfig, csCache);
+  public ConnectionFactory(final HostAndPort hostAndPort, final JedisClientConfig clientConfig, Cache csCache) {
+    this(builder().hostAndPort(hostAndPort).clientConfig(clientConfig).cache(csCache).withDefaults());
   }
 
-  public ConnectionFactory(final JedisSocketFactory jedisSocketFactory,
-      final JedisClientConfig clientConfig) {
-    this(jedisSocketFactory, clientConfig, null);
+  public ConnectionFactory(final JedisSocketFactory jedisSocketFactory, final JedisClientConfig clientConfig) {
+    this(builder().socketFactory(jedisSocketFactory).clientConfig(clientConfig).withDefaults());
   }
 
-  private ConnectionFactory(final JedisSocketFactory jedisSocketFactory,
-      final JedisClientConfig clientConfig, Cache csCache) {
+  public ConnectionFactory(Builder builder) {
+    this.clientConfig = builder.getClientConfig();
+    this.connectionBuilder = builder.getConnectionBuilder();
 
-    this.jedisSocketFactory = jedisSocketFactory;
-    this.clientSideCache = csCache;
-    this.clientConfig = clientConfig;
+    initAuthXManager();
+  }
 
+  private void initAuthXManager() {
     AuthXManager authXManager = clientConfig.getAuthXManager();
     if (authXManager == null) {
-      this.objectMaker = connectionSupplier();
+      this.objectMaker = () -> build();
       this.authXEventListener = AuthXEventListener.NOOP_LISTENER;
     } else {
-      Supplier<Connection> supplier = connectionSupplier();
-      this.objectMaker = () -> (Connection) authXManager.addConnection(supplier.get());
+      this.objectMaker = () -> (Connection) authXManager.addConnection(build());
       this.authXEventListener = authXManager.getListener();
       authXManager.start();
     }
   }
 
-  private Supplier<Connection> connectionSupplier() {
-    return clientSideCache == null ? () -> new Connection(jedisSocketFactory, clientConfig)
-        : () -> new CacheConnection(jedisSocketFactory, clientConfig, clientSideCache);
+  private Connection build() {
+    return connectionBuilder.build();
   }
 
   @Override

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

@@ -3,6 +3,7 @@
 import org.apache.commons.pool2.PooledObjectFactory;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 
+import redis.clients.authentication.core.Token;
 import redis.clients.jedis.annots.Experimental;
 import redis.clients.jedis.authentication.AuthXManager;
 import redis.clients.jedis.csc.Cache;
@@ -13,25 +14,32 @@ public class ConnectionPool extends Pool<Connection> {
 
   private AuthXManager authXManager;
 
+  // Primary constructors using factory
+  public ConnectionPool(PooledObjectFactory<Connection> factory) {
+    super(factory);
+  }
+
+  public ConnectionPool(PooledObjectFactory<Connection> factory,
+      GenericObjectPoolConfig<Connection> poolConfig) {
+    super(factory, poolConfig);
+  }
+
+  // Convenience constructors
   public ConnectionPool(HostAndPort hostAndPort, JedisClientConfig clientConfig) {
     this(new ConnectionFactory(hostAndPort, clientConfig));
     attachAuthenticationListener(clientConfig.getAuthXManager());
   }
 
-  @Experimental
   public ConnectionPool(HostAndPort hostAndPort, JedisClientConfig clientConfig,
-      Cache clientSideCache) {
-    this(new ConnectionFactory(hostAndPort, clientConfig, clientSideCache));
+      GenericObjectPoolConfig<Connection> poolConfig) {
+    this(new ConnectionFactory(hostAndPort, clientConfig), poolConfig);
     attachAuthenticationListener(clientConfig.getAuthXManager());
   }
 
-  public ConnectionPool(PooledObjectFactory<Connection> factory) {
-    super(factory);
-  }
-
+  @Experimental
   public ConnectionPool(HostAndPort hostAndPort, JedisClientConfig clientConfig,
-      GenericObjectPoolConfig<Connection> poolConfig) {
-    this(new ConnectionFactory(hostAndPort, clientConfig), poolConfig);
+      Cache clientSideCache) {
+    this(new ConnectionFactory(hostAndPort, clientConfig, clientSideCache));
     attachAuthenticationListener(clientConfig.getAuthXManager());
   }
 
@@ -42,11 +50,6 @@ public ConnectionPool(HostAndPort hostAndPort, JedisClientConfig clientConfig,
     attachAuthenticationListener(clientConfig.getAuthXManager());
   }
 
-  public ConnectionPool(PooledObjectFactory<Connection> factory,
-      GenericObjectPoolConfig<Connection> poolConfig) {
-    super(factory, poolConfig);
-  }
-
   @Override
   public Connection getResource() {
     Connection conn = super.getResource();
@@ -65,17 +68,25 @@ public void close() {
     }
   }
 
-  private void attachAuthenticationListener(AuthXManager authXManager) {
+  protected void attachAuthenticationListener(AuthXManager authXManager) {
     this.authXManager = authXManager;
     if (authXManager != null) {
-      authXManager.addPostAuthenticationHook(token -> {
-        try {
-          // this is to trigger validations on each connection via ConnectionFactory
-          evict();
-        } catch (Exception e) {
-          throw new JedisException("Failed to evict connections from pool", e);
-        }
-      });
+      authXManager.addPostAuthenticationHook(this::postAuthentication);
+    }
+  }
+
+  protected void detachAuthenticationListener() {
+    if (authXManager != null) {
+      authXManager.removePostAuthenticationHook(this::postAuthentication);
+    }
+  }
+
+  private void postAuthentication(Token token) {
+    try {
+      // this is to trigger validations on each connection via ConnectionFactory
+      evict();
+    } catch (Exception e) {
+      throw new JedisException("Failed to evict connections from pool", e);
     }
   }
 }