[automatic-failover] Integrate health checks with probing policies and retry logic (CAE-1685) (#3541)

* initial port Jedis health monitoring

* wip integrate healthchecks

* formating

* formating

* add test case plan

* Endpoints without health checks configured should return HEALTHY

Changes
 - add connection.getHealthStatus(RedisUri endpoint)
 - HEALTHY - returned for Databases without health checks configured
 - add test

* Create MultiDbClient with custom health check strategy supplier

Changes
 - add test to ensure health status changes from custom health checks are reflected

* Create MultiDbClient with custom health check strategy supplier

Changes
 - add test to ensure health status changes from custom health checks are reflected

* faster await timeout

* add test - use different health check strategies for different endpoints

* wait for initial healthy database

* add test - configure health check interval and timeout

* add test - trigger failover when health check detects unhealthy endpoint

* add test - should not failover to unhealthy endpoints

* add test - Should trigger failover via circuit breaker even when health check returns HEALTHY

* reduce await poll interval in HealthCheckIntegrationTest

* mark un-implemented tests are disabled

* add test - Should transition from UNKNOWN to HEALTHY

* add test - Should create health check when adding new database

* fix - Should stop health check when removing  database

* add test - Should stop health check when removing  database

* add test - HealthCheckLifecycleTests
  - Should start health checks automatically when connection is created
  - Should stop health checks when connection is closed

* fix HealthCheck not stopped on StatefulRedisMultiDbConnection.close()

* remove HealthStatusListenerTests stubs, health check events, not exposed publicly

* format

* add health checks unit test

* clean up

   - rename health check thread names to lettuce-*
   - clean up warnings
   - format
   - javadocs & autor updated

* address failing tests

  - Update  StatefulMultiDbConnectionIntegrationTests to account for added additional test server in MultiDbTestSupport
  - Junit4  @after replaced with JUnit5

* address failing tests

  - Update  StatefulMultiDbConnectionIntegrationTests to account for added additional test server in MultiDbTestSupport
  - Junit4  @after replaced with JUnit5

* package private StatusTracker

* make healthStatusManager required when creating MultiDbStatefullConnection

* remove un-implemented probing integration tests

 - covered with unit tests

* introduce isHealthy() to replace getHealthStatus()

* register listeners before adding HealthChecks

Author: ggivo

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

@@ -10,6 +10,7 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 
+import io.lettuce.core.internal.LettuceAssert;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -46,6 +47,8 @@ public class CircuitBreaker implements Closeable {
      * Create a circuit breaker instance.
      */
     public CircuitBreaker(CircuitBreakerConfig config) {
+        LettuceAssert.notNull(config, "CircuitBreakerConfig must not be null");
+
         this.config = config;
         this.trackedExceptions = new HashSet<>(config.trackedExceptions);
         this.stateRef = new AtomicReference<>(

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

@@ -3,10 +3,12 @@
 import io.lettuce.core.ClientOptions;
 import io.lettuce.core.RedisURI;
 import io.lettuce.core.failover.CircuitBreaker.CircuitBreakerConfig;
+import io.lettuce.core.failover.health.HealthCheckStrategySupplier;
 import io.lettuce.core.internal.LettuceAssert;
 
 /**
- * Configuration for a database in a multi-database client. Holds the Redis URI, weight for load balancing, and client options.
+ * Configuration for a database in a multi-database client. Holds the Redis URI, weight for load balancing, client options,
+ * circuit breaker configuration, and optional health check strategy supplier.
  *
  * @author Ali Takavci
  * @since 7.1
@@ -21,44 +23,61 @@ public class DatabaseConfig {
 
     private final CircuitBreakerConfig circuitBreakerConfig;
 
+    private final HealthCheckStrategySupplier healthCheckStrategySupplier;
+
     /**
      * Create a new database configuration.
      *
      * @param redisURI the Redis URI, must not be {@code null}
      * @param weight the weight for load balancing, must be greater than 0
      * @param clientOptions the client options, can be {@code null} to use defaults
      * @param circuitBreakerConfig the circuit breaker configuration, can be {@code null} to use defaults
+     * @param healthCheckStrategySupplier the health check strategy supplier, can be {@code null} to disable health checks
      */
     public DatabaseConfig(RedisURI redisURI, float weight, ClientOptions clientOptions,
-            CircuitBreakerConfig circuitBreakerConfig) {
+            CircuitBreakerConfig circuitBreakerConfig, HealthCheckStrategySupplier healthCheckStrategySupplier) {
         LettuceAssert.notNull(redisURI, "RedisURI must not be null");
         LettuceAssert.isTrue(weight > 0, "Weight must be greater than 0");
 
         this.redisURI = redisURI;
         this.weight = weight;
         this.clientOptions = clientOptions;
         this.circuitBreakerConfig = circuitBreakerConfig != null ? circuitBreakerConfig : CircuitBreakerConfig.DEFAULT;
+        this.healthCheckStrategySupplier = healthCheckStrategySupplier;
+    }
+
+    /**
+     * Create a new database configuration with default health check strategy supplier.
+     *
+     * @param redisURI the Redis URI, must not be {@code null}
+     * @param weight the weight for load balancing, must be greater than 0
+     * @param clientOptions the client options, can be {@code null} to use defaults
+     * @param circuitBreakerConfig the circuit breaker configuration, can be {@code null} to use defaults
+     */
+    public DatabaseConfig(RedisURI redisURI, float weight, ClientOptions clientOptions,
+            CircuitBreakerConfig circuitBreakerConfig) {
+        this(redisURI, weight, clientOptions, circuitBreakerConfig, null);
     }
 
     /**
-     * Create a new database configuration with default client options.
+     * Create a new database configuration with default client options and health check strategy supplier.
      *
      * @param redisURI the Redis URI, must not be {@code null}
      * @param weight the weight for load balancing, must be greater than 0
      * @param clientOptions the client options, can be {@code null} to use defaults
      */
     public DatabaseConfig(RedisURI redisURI, float weight, ClientOptions clientOptions) {
-        this(redisURI, weight, clientOptions, null);
+        this(redisURI, weight, clientOptions, null, null);
     }
 
     /**
-     * Create a new database configuration with default client options.
+     * Create a new database configuration with default client options and health check strategy supplier.
      *
      * @param redisURI the Redis URI, must not be {@code null}
      * @param weight the weight for load balancing, must be greater than 0
      */
     public DatabaseConfig(RedisURI redisURI, float weight) {
-        this(redisURI, weight, null, null);
+        this(redisURI, weight, null, null, null);
     }
 
     /**
@@ -97,6 +116,15 @@ public CircuitBreakerConfig getCircuitBreakerConfig() {
         return circuitBreakerConfig;
     }
 
+    /**
+     * Get the health check strategy supplier.
+     *
+     * @return the health check strategy supplier, can be {@code null}
+     */
+    public HealthCheckStrategySupplier getHealthCheckStrategySupplier() {
+        return healthCheckStrategySupplier;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o)
@@ -110,20 +138,30 @@ public boolean equals(Object o) {
             return false;
         if (!redisURI.equals(that.redisURI))
             return false;
-        return clientOptions != null ? clientOptions.equals(that.clientOptions) : that.clientOptions == null;
+        if (clientOptions != null ? !clientOptions.equals(that.clientOptions) : that.clientOptions != null)
+            return false;
+        if (circuitBreakerConfig != null ? !circuitBreakerConfig.equals(that.circuitBreakerConfig)
+                : that.circuitBreakerConfig != null)
+            return false;
+        return healthCheckStrategySupplier != null ? healthCheckStrategySupplier.equals(that.healthCheckStrategySupplier)
+                : that.healthCheckStrategySupplier == null;
     }
 
     @Override
     public int hashCode() {
         int result = redisURI.hashCode();
         result = 31 * result + (weight != +0.0f ? Float.floatToIntBits(weight) : 0);
         result = 31 * result + (clientOptions != null ? clientOptions.hashCode() : 0);
+        result = 31 * result + (circuitBreakerConfig != null ? circuitBreakerConfig.hashCode() : 0);
+        result = 31 * result + (healthCheckStrategySupplier != null ? healthCheckStrategySupplier.hashCode() : 0);
         return result;
     }
 
     @Override
     public String toString() {
-        return "DatabaseConfig{" + "redisURI=" + redisURI + ", weight=" + weight + ", clientOptions=" + clientOptions + '}';
+        return "DatabaseConfig{" + "redisURI=" + redisURI + ", weight=" + weight + ", clientOptions=" + clientOptions
+                + ", circuitBreakerConfig=" + circuitBreakerConfig + ", healthCheckStrategySupplier="
+                + healthCheckStrategySupplier + '}';
     }
 
 }

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

@@ -2,6 +2,7 @@
 
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.codec.RedisCodec;
+import io.lettuce.core.failover.health.HealthStatusManager;
 
 /**
  * Factory interface for creating database connections in a multi-database client.
@@ -23,6 +24,6 @@ interface DatabaseConnectionFactory<C extends StatefulRedisConnection<K, V>, K,
      * @param codec the codec to use for encoding/decoding
      * @return a new RedisDatabase instance
      */
-    RedisDatabase<C> createDatabase(DatabaseConfig config, RedisCodec<K, V> codec);
+    RedisDatabase<C> createDatabase(DatabaseConfig config, RedisCodec<K, V> codec, HealthStatusManager healthStatusManager);
 
 }

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

@@ -1,23 +1,34 @@
 package io.lettuce.core.failover;
 
 import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
 
 import io.lettuce.core.Delegating;
 import io.lettuce.core.RedisChannelWriter;
 import io.lettuce.core.RedisClient;
+import io.lettuce.core.RedisConnectionException;
 import io.lettuce.core.RedisURI;
 import io.lettuce.core.StatefulRedisConnectionImpl;
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.codec.RedisCodec;
 import io.lettuce.core.failover.api.StatefulRedisMultiDbConnection;
 import io.lettuce.core.failover.api.StatefulRedisMultiDbPubSubConnection;
+import io.lettuce.core.failover.health.HealthCheck;
+import io.lettuce.core.failover.health.HealthCheckStrategy;
+import io.lettuce.core.failover.health.HealthStatus;
+import io.lettuce.core.failover.health.HealthStatusManager;
+import io.lettuce.core.failover.health.HealthStatusManagerImpl;
 import io.lettuce.core.internal.LettuceAssert;
 import io.lettuce.core.protocol.DefaultEndpoint;
 import io.lettuce.core.pubsub.PubSubEndpoint;
 import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
 import io.lettuce.core.resource.ClientResources;
+import io.netty.util.internal.logging.InternalLogger;
+import io.netty.util.internal.logging.InternalLoggerFactory;
 
 /**
  * Failover-aware client that composes multiple standalone Redis endpoints and returns a single Stateful connection wrapper
@@ -30,6 +41,8 @@
  */
 class MultiDbClientImpl extends RedisClient implements MultiDbClient {
 
+    private static final InternalLogger logger = InternalLoggerFactory.getInstance(MultiDbClientImpl.class);
+
     private static final RedisURI EMPTY_URI = new RedisURI();
 
     private final Map<RedisURI, DatabaseConfig> databaseConfigs;
@@ -73,6 +86,8 @@ public <K, V> StatefulRedisMultiDbConnection<K, V> connect(RedisCodec<K, V> code
             throw new IllegalArgumentException("codec must not be null");
         }
 
+        HealthStatusManager healthStatusManager = createHealthStatusManager();
+
         Map<RedisURI, RedisDatabase<StatefulRedisConnection<K, V>>> databases = new ConcurrentHashMap<>(databaseConfigs.size());
         for (Map.Entry<RedisURI, DatabaseConfig> entry : databaseConfigs.entrySet()) {
             RedisURI uri = entry.getKey();
@@ -81,22 +96,41 @@ public <K, V> StatefulRedisMultiDbConnection<K, V> connect(RedisCodec<K, V> code
             // HACK: looks like repeating the implementation all around 'RedisClient.connect' is an overkill.
             // connections.put(uri, connect(codec, uri));
             // Instead we will use it from delegate
-            RedisDatabase<StatefulRedisConnection<K, V>> database = createRedisDatabase(config, codec);
+            RedisDatabase<StatefulRedisConnection<K, V>> database = createRedisDatabase(config, codec, healthStatusManager);
 
             databases.put(uri, database);
         }
 
+        StatusTracker statusTracker = new StatusTracker(healthStatusManager);
+        // Wait for health checks to complete if configured
+        waitForInitialHealthyDatabase(statusTracker, databases);
+
         // Provide a connection factory for dynamic database addition
         return new StatefulRedisMultiDbConnectionImpl<StatefulRedisConnection<K, V>, K, V>(databases, getResources(), codec,
-                getOptions().getJsonParser(), this::createRedisDatabase);
+                getOptions().getJsonParser(), this::createRedisDatabase, healthStatusManager);
+    }
+
+    protected HealthStatusManager createHealthStatusManager() {
+        return new HealthStatusManagerImpl();
     }
 
     private <K, V> RedisDatabase<StatefulRedisConnection<K, V>> createRedisDatabase(DatabaseConfig config,
-            RedisCodec<K, V> codec) {
+            RedisCodec<K, V> codec, HealthStatusManager healthStatusManager) {
         RedisURI uri = config.getRedisURI();
         StatefulRedisConnection<K, V> connection = connect(codec, uri);
         DatabaseEndpoint databaseEndpoint = extractDatabaseEndpoint(connection);
-        RedisDatabase<StatefulRedisConnection<K, V>> database = new RedisDatabase<>(config, connection, databaseEndpoint);
+
+        HealthCheck healthCheck;
+        if (config.getHealthCheckStrategySupplier() != null) {
+            HealthCheckStrategy hcStrategy = config.getHealthCheckStrategySupplier().get(config.getRedisURI(),
+                    connection.getOptions());
+            healthCheck = healthStatusManager.add(uri, hcStrategy);
+        } else {
+            healthCheck = null;
+        }
+
+        RedisDatabase<StatefulRedisConnection<K, V>> database = new RedisDatabase<>(config, connection, databaseEndpoint,
+                healthCheck);
 
         return database;
     }
@@ -116,27 +150,45 @@ public <K, V> StatefulRedisMultiDbPubSubConnection<K, V> connectPubSub(RedisCode
             throw new IllegalArgumentException("codec must not be null");
         }
 
+        HealthStatusManager healthStatusManager = createHealthStatusManager();
+
         Map<RedisURI, RedisDatabase<StatefulRedisPubSubConnection<K, V>>> databases = new ConcurrentHashMap<>(
                 databaseConfigs.size());
         for (Map.Entry<RedisURI, DatabaseConfig> entry : databaseConfigs.entrySet()) {
             RedisURI uri = entry.getKey();
             DatabaseConfig config = entry.getValue();
 
-            RedisDatabase<StatefulRedisPubSubConnection<K, V>> database = createRedisDatabaseWithPubSub(config, codec);
+            RedisDatabase<StatefulRedisPubSubConnection<K, V>> database = createRedisDatabaseWithPubSub(config, codec,
+                    healthStatusManager);
             databases.put(uri, database);
         }
 
+        StatusTracker statusTracker = new StatusTracker(healthStatusManager);
+        // Wait for health checks to complete if configured
+        waitForInitialHealthyDatabase(statusTracker, databases);
+
         // Provide a connection factory for dynamic database addition
         return new StatefulRedisMultiDbPubSubConnectionImpl<K, V>(databases, getResources(), codec,
-                getOptions().getJsonParser(), this::createRedisDatabaseWithPubSub);
+                getOptions().getJsonParser(), this::createRedisDatabaseWithPubSub, healthStatusManager);
     }
 
     private <K, V> RedisDatabase<StatefulRedisPubSubConnection<K, V>> createRedisDatabaseWithPubSub(DatabaseConfig config,
-            RedisCodec<K, V> codec) {
+            RedisCodec<K, V> codec, HealthStatusManager healthStatusManager) {
         RedisURI uri = config.getRedisURI();
         StatefulRedisPubSubConnection<K, V> connection = connectPubSub(codec, uri);
         DatabaseEndpoint databaseEndpoint = extractDatabaseEndpoint(connection);
-        RedisDatabase<StatefulRedisPubSubConnection<K, V>> database = new RedisDatabase<>(config, connection, databaseEndpoint);
+
+        HealthCheck healthCheck;
+        if (config.getHealthCheckStrategySupplier() != null) {
+            HealthCheckStrategy hcStrategy = config.getHealthCheckStrategySupplier().get(config.getRedisURI(),
+                    connection.getOptions());
+            healthCheck = healthStatusManager.add(uri, hcStrategy);
+        } else {
+            healthCheck = null;
+        }
+
+        RedisDatabase<StatefulRedisPubSubConnection<K, V>> database = new RedisDatabase<>(config, connection, databaseEndpoint,
+                healthCheck);
         return database;
     }
 
@@ -158,4 +210,52 @@ protected <K, V> PubSubEndpoint<K, V> createPubSubEndpoint() {
         return new DatabasePubSubEndpointImpl<>(getOptions(), getResources());
     }
 
+    /**
+     * Waits for initial health check results and selects the first healthy database based on weight priority. Blocks until at
+     * least one database becomes healthy or all databases are determined to be unhealthy.
+     *
+     * @param statusTracker the status tracker to use for waiting on health check results
+     * @param databaseMap the map of databases to evaluate
+     * @throws RedisConnectionException if all databases are unhealthy
+     */
+    private void waitForInitialHealthyDatabase(StatusTracker statusTracker,
+            Map<RedisURI, ? extends RedisDatabase<?>> databaseMap) {
+        // Sort databases by weight in descending order
+        List<? extends Map.Entry<RedisURI, ? extends RedisDatabase<?>>> sortedDatabases = databaseMap.entrySet().stream()
+                .sorted(Map.Entry.comparingByValue(Comparator.comparing((RedisDatabase<?> db) -> db.getWeight()).reversed()))
+                .collect(Collectors.toList());
+        logger.info("Selecting initial database from {} configured databases", sortedDatabases.size());
+
+        // Select database in weight order
+        for (Map.Entry<RedisURI, ? extends RedisDatabase<?>> entry : sortedDatabases) {
+            RedisURI endpoint = entry.getKey();
+            RedisDatabase<?> database = entry.getValue();
+
+            logger.info("Evaluating database {} (weight: {})", endpoint, database.getWeight());
+
+            HealthStatus status;
+
+            // Check if health checks are enabled for this database
+            if (database.getHealthCheck() != null) {
+                logger.info("Health checks enabled for {}, waiting for result", endpoint);
+                // Wait for this database's health status to be determined
+                status = statusTracker.waitForHealthStatus(endpoint);
+            } else {
+                // No health check configured - assume healthy
+                logger.info("No health check configured for database {}, defaulting to HEALTHY", endpoint);
+                status = HealthStatus.HEALTHY;
+            }
+
+            if (status == HealthStatus.HEALTHY) {
+                logger.info("Found healthy database: {} (weight: {})", endpoint, database.getWeight());
+                return;
+            } else {
+                logger.info("Database {} is unhealthy, trying next database", endpoint);
+            }
+        }
+
+        // All databases are unhealthy
+        throw new RedisConnectionException("All configured databases are unhealthy.");
+    }
+
 }