[automatic failover] Introduce fast failover mode - a thread-sync-free approach with builders

.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);
     }
   }
 }