[automatic failover] Implement wait on healthCheck results during client init (#4207)

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

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

---------

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

Author: atakavci

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

@@ -162,6 +162,12 @@ public static interface StrategySupplier {
     private long gracePeriod;
 
     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;
     }
 

src/main/java/redis/clients/jedis/mcf/HealthCheck.java

@@ -27,7 +27,7 @@ public class HealthCheck {
         this.endpoint = endpoint;
         this.strategy = strategy;
         this.statusChangeCallback = statusChangeCallback;
-        statusRef.set(new SimpleEntry<>(0L, HealthStatus.HEALTHY));
+        statusRef.set(new SimpleEntry<>(0L, HealthStatus.UNKNOWN));
     }
 
     public Endpoint getEndpoint() {
@@ -92,9 +92,9 @@ private void safeUpdate(long owner, HealthStatus status) {
             }
             return current;
         });
-        if (oldStatus.getKey() != owner || oldStatus.getValue() != status) {
+        if (oldStatus.getValue() != status) {
             // notify listeners
-            notifyListeners(oldStatus.getValue(), newStatus.getValue());
+            notifyListeners(oldStatus.getValue(), status);
         }
     }
 

src/main/java/redis/clients/jedis/mcf/HealthStatus.java

@@ -1,7 +1,7 @@
 package redis.clients.jedis.mcf;
 
 public enum HealthStatus {
-    HEALTHY(0x01), UNHEALTHY(0x02);
+    UNKNOWN(0x00), HEALTHY(0x01), UNHEALTHY(0x02);
 
     private final int value;
 

src/main/java/redis/clients/jedis/mcf/HealthStatusManager.java

@@ -72,6 +72,10 @@ public void removeAll(Endpoint[] endpoints) {
 
     public HealthStatus getHealthStatus(Endpoint endpoint) {
         HealthCheck healthCheck = healthChecks.get(endpoint);
-        return healthCheck != null ? healthCheck.getStatus() : HealthStatus.UNHEALTHY;
+        return healthCheck != null ? healthCheck.getStatus() : HealthStatus.UNKNOWN;
+    }
+
+    public boolean hasHealthCheck(Endpoint endpoint) {
+        return healthChecks.get(endpoint) != null;
     }
 }

src/main/java/redis/clients/jedis/mcf/StatusTracker.java

@@ -0,0 +1,77 @@
+package redis.clients.jedis.mcf;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import redis.clients.jedis.exceptions.JedisConnectionException;
+import redis.clients.jedis.exceptions.JedisValidationException;
+
+/**
+ * StatusTracker is responsible for tracking and waiting for health status changes for specific endpoints. It provides
+ * an event-driven approach to wait for health status transitions from UNKNOWN to either HEALTHY or UNHEALTHY.
+ */
+public class StatusTracker {
+
+    private final HealthStatusManager healthStatusManager;
+
+    public StatusTracker(HealthStatusManager healthStatusManager) {
+        this.healthStatusManager = healthStatusManager;
+    }
+
+    /**
+     * Waits for a specific endpoint's health status to be determined (not UNKNOWN). Uses event-driven approach with
+     * CountDownLatch to avoid polling.
+     * @param endpoint the endpoint to wait for
+     * @return the determined health status (HEALTHY or UNHEALTHY)
+     * @throws JedisConnectionException if interrupted while waiting
+     */
+    public HealthStatus waitForHealthStatus(Endpoint endpoint) {
+        // First check if status is already determined
+        HealthStatus currentStatus = healthStatusManager.getHealthStatus(endpoint);
+        if (currentStatus != HealthStatus.UNKNOWN) {
+            return currentStatus;
+        }
+
+        // Set up event-driven waiting
+        final CountDownLatch latch = new CountDownLatch(1);
+        final AtomicReference<HealthStatus> resultStatus = new AtomicReference<>();
+
+        // Create a temporary listener for this specific endpoint
+        HealthStatusListener tempListener = new HealthStatusListener() {
+            @Override
+            public void onStatusChange(HealthStatusChangeEvent event) {
+                if (event.getEndpoint().equals(endpoint) && event.getNewStatus() != HealthStatus.UNKNOWN) {
+                    resultStatus.set(event.getNewStatus());
+                    latch.countDown();
+                }
+            }
+        };
+
+        // Register the temporary listener
+        healthStatusManager.registerListener(endpoint, tempListener);
+
+        try {
+            // Double-check status after registering listener (race condition protection)
+            currentStatus = healthStatusManager.getHealthStatus(endpoint);
+            if (currentStatus != HealthStatus.UNKNOWN) {
+                return currentStatus;
+            }
+
+            // Wait for the health status change event
+            // just for safety to not block indefinitely
+            boolean completed = latch.await(60, TimeUnit.SECONDS);
+            if (!completed) {
+                throw new JedisValidationException("Timeout while waiting for health check result");
+            }
+            return resultStatus.get();
+
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new JedisConnectionException("Interrupted while waiting for health check result", e);
+        } finally {
+            // Clean up: unregister the temporary listener
+            healthStatusManager.unregisterListener(endpoint, tempListener);
+        }
+    }
+}