[automatic-failover] Implement weighted endpoint selection (#3519)

* -  drop RedisDatabaseConfig

- add HealthStatus

* - set default healthStatus.Healthy

* - review from @ggivo, improve readability

* review from @tishun
- rename predicate function
- add javadoc

* - format

* - fix java docs

* - format

Author: atakavci

src/main/java/io/lettuce/core/failover/HealthStatus.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2011-Present, Redis Ltd. and Contributors
+ * All rights reserved.
+ *
+ * Licensed under the MIT License.
+ *
+ * This file contains contributions from third-party contributors
+ * licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.lettuce.core.failover;
+
+/**
+ * Enumeration representing the health status of a database endpoint as determined by the health check service.
+ * <p>
+ * This status reflects the results from active health checks performed by the health check service. Health checks periodically
+ * probes endpoints to verify their availability and responsiveness, updating this status accordingly.
+ * <p>
+ * Used by the {@link RedisDatabase} to track and report health check results from the health check API.
+ *
+ * @author Ali Takavci
+ * @since 7.1
+ */
+public enum HealthStatus {
+
+    /**
+     * Health status is unknown. This is the initial state before any health checks have been performed by the health check
+     * service.
+     */
+    UNKNOWN,
+
+    /**
+     * The endpoint is healthy and operating normally according to health check probes. Health checks are passing and the
+     * endpoint is responding as expected.
+     */
+    HEALTHY,
+
+    /**
+     * The endpoint is unhealthy according to health check probes. Health checks are failing, indicating the endpoint is not
+     * responding or not meeting health criteria.
+     */
+    UNHEALTHY
+
+}

src/main/java/io/lettuce/core/failover/MultiDbClientImpl.java

@@ -96,9 +96,7 @@ private <K, V> RedisDatabase<StatefulRedisConnection<K, V>> createRedisDatabase(
         RedisURI uri = config.getRedisURI();
         StatefulRedisConnection<K, V> connection = connect(codec, uri);
         DatabaseEndpoint databaseEndpoint = extractDatabaseEndpoint(connection);
-        RedisDatabase<StatefulRedisConnection<K, V>> database = new RedisDatabase<>(
-                new RedisDatabase.RedisDatabaseConfig(uri, config.getWeight(), config.getCircuitBreakerConfig()), connection,
-                databaseEndpoint);
+        RedisDatabase<StatefulRedisConnection<K, V>> database = new RedisDatabase<>(config, connection, databaseEndpoint);
 
         return database;
     }
@@ -138,9 +136,7 @@ private <K, V> RedisDatabase<StatefulRedisPubSubConnection<K, V>> createRedisDat
         RedisURI uri = config.getRedisURI();
         StatefulRedisPubSubConnection<K, V> connection = connectPubSub(codec, uri);
         DatabaseEndpoint databaseEndpoint = extractDatabaseEndpoint(connection);
-        RedisDatabase<StatefulRedisPubSubConnection<K, V>> database = new RedisDatabase<>(
-                new RedisDatabase.RedisDatabaseConfig(uri, config.getWeight(), config.getCircuitBreakerConfig()), connection,
-                databaseEndpoint);
+        RedisDatabase<StatefulRedisPubSubConnection<K, V>> database = new RedisDatabase<>(config, connection, databaseEndpoint);
         return database;
     }
 

src/main/java/io/lettuce/core/failover/RedisDatabase.java

@@ -4,7 +4,6 @@
 
 import io.lettuce.core.RedisURI;
 import io.lettuce.core.api.StatefulRedisConnection;
-import io.lettuce.core.failover.CircuitBreaker.CircuitBreakerConfig;
 
 /**
  * Represents a Redis database with a weight and a connection.
@@ -26,12 +25,14 @@ public class RedisDatabase<C extends StatefulRedisConnection<?, ?>> implements C
 
     private final CircuitBreaker circuitBreaker;
 
-    public RedisDatabase(RedisDatabaseConfig config, C connection, DatabaseEndpoint databaseEndpoint) {
-        this.redisURI = config.redisURI;
-        this.weight = config.weight;
+    private HealthStatus healthStatus = HealthStatus.HEALTHY;
+
+    public RedisDatabase(DatabaseConfig config, C connection, DatabaseEndpoint databaseEndpoint) {
+        this.redisURI = config.getRedisURI();
+        this.weight = config.getWeight();
         this.connection = connection;
         this.databaseEndpoint = databaseEndpoint;
-        this.circuitBreaker = new CircuitBreaker(config.circuitBreakerConfig);
+        this.circuitBreaker = new CircuitBreaker(config.getCircuitBreakerConfig());
         databaseEndpoint.setCircuitBreaker(circuitBreaker);
     }
 
@@ -61,26 +62,8 @@ public void close() {
         circuitBreaker.close();
     }
 
-    public static class RedisDatabaseConfig {
-
-        private RedisURI redisURI;
-
-        private float weight;
-
-        private CircuitBreakerConfig circuitBreakerConfig;
-
-        public RedisDatabaseConfig(RedisURI redisURI, float weight) {
-            this.redisURI = redisURI;
-            this.weight = weight;
-            this.circuitBreakerConfig = CircuitBreakerConfig.DEFAULT;
-        }
-
-        public RedisDatabaseConfig(RedisURI redisURI, float weight, CircuitBreakerConfig circuitBreakerConfig) {
-            this.redisURI = redisURI;
-            this.weight = weight;
-            this.circuitBreakerConfig = circuitBreakerConfig;
-        }
-
+    public HealthStatus getHealthStatus() {
+        return healthStatus;
     }
 
 }

src/main/java/io/lettuce/core/failover/StatefulRedisMultiDbConnectionImpl.java

@@ -11,6 +11,7 @@
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
 
 import io.lettuce.core.AbstractRedisClient;
@@ -46,6 +47,7 @@ public class StatefulRedisMultiDbConnectionImpl<C extends StatefulRedisConnectio
 
     protected final Map<RedisURI, RedisDatabase<C>> databases;
 
+    // this should not be null ever after succesfull initialization
     protected RedisDatabase<C> current;
 
     protected final RedisCommands<K, V> sync;
@@ -79,7 +81,11 @@ public StatefulRedisMultiDbConnectionImpl(Map<RedisURI, RedisDatabase<C>> connec
         this.codec = codec;
         this.parser = parser;
         this.connectionFactory = connectionFactory;
-        this.current = connections.values().stream().max(Comparator.comparingDouble(RedisDatabase::getWeight)).get();
+        // TODO: Current implementation forces all database connections to be created and established (at least once before this
+        // constructor called).
+        // This is suboptimal and should be replaced with a logic that uses async connection creation and state management,
+        // which safely starts with at least one healthy connection.
+        this.current = getNextHealthyDatabase(null);
 
         this.async = newRedisAsyncCommandsImpl();
         this.sync = newRedisSyncCommandsImpl();
@@ -95,7 +101,7 @@ private void onCircuitBreakerStateChange(CircuitBreakerStateChangeEvent event) {
     }
 
     private void failoverFrom(RedisDatabase<C> fromDb) {
-        RedisDatabase<C> healthyDatabase = getHealthyDatabase(fromDb);
+        RedisDatabase<C> healthyDatabase = getNextHealthyDatabase(fromDb);
         if (healthyDatabase != null) {
             switchToDatabase(healthyDatabase.getRedisURI());
         } else {
@@ -104,10 +110,25 @@ private void failoverFrom(RedisDatabase<C> fromDb) {
         }
     }
 
-    private RedisDatabase<C> getHealthyDatabase(RedisDatabase<C> current) {
-        return databases.values().stream().filter(db -> db != current)
-                .filter(db -> db.getCircuitBreaker().getCurrentState() == CircuitBreaker.State.CLOSED)
-                .max(Comparator.comparingDouble(RedisDatabase::getWeight)).get();
+    private RedisDatabase<C> getNextHealthyDatabase(RedisDatabase<C> dbToExclude) {
+        return databases.values().stream().filter(DatabasePredicates.isHealthy).filter(DatabasePredicates.isNot(dbToExclude))
+                .max(DatabaseComparators.byWeight).orElse(null);
+    }
+
+    static class DatabaseComparators {
+
+        public static final Comparator<RedisDatabase<?>> byWeight = Comparator.comparingDouble(RedisDatabase::getWeight);
+
+    }
+
+    static class DatabasePredicates {
+
+        public static final Predicate<RedisDatabase<?>> isHealthy = db -> db.getHealthStatus() == HealthStatus.HEALTHY;
+
+        public static Predicate<RedisDatabase<?>> isNot(RedisDatabase<?> dbInstance) {
+            return db -> !db.equals(dbInstance);
+        }
+
     }
 
     @Override