- adressing reviews and feedback

Author: atakavci

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

@@ -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/StatusTracker.java

@@ -1,9 +1,11 @@
 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
@@ -57,7 +59,11 @@ public void onStatusChange(HealthStatusChangeEvent event) {
             }
 
             // Wait for the health status change event
-            latch.await();
+            // 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) {

src/main/java/redis/clients/jedis/providers/MultiClusterPooledConnectionProvider.java

@@ -15,6 +15,7 @@
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
@@ -97,8 +98,9 @@ public class MultiClusterPooledConnectionProvider implements ConnectionProvider
     private final ConcurrentLinkedQueue<HealthStatusChangeEvent> pendingHealthStatusChanges = new ConcurrentLinkedQueue<>();
 
     // Failback mechanism fields
+    private static final AtomicInteger failbackThreadCounter = new AtomicInteger(1);
     private final ScheduledExecutorService failbackScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
-        Thread t = new Thread(r, "failback-scheduler");
+        Thread t = new Thread(r, "jedis-failback-" + failbackThreadCounter.getAndIncrement());
         t.setDaemon(true);
         return t;
     });
@@ -333,7 +335,11 @@ private void processPendingHealthStatusChanges() {
         synchronized (pendingHealthStatusChanges) {
             // Process all queued events
             while ((event = pendingHealthStatusChanges.poll()) != null) {
-                processStatusChangeEvent(event);
+                Endpoint endpoint = event.getEndpoint();
+                boolean latestInTheQueue = !pendingHealthStatusChanges.stream().anyMatch(e -> e.getEndpoint().equals(endpoint));
+                if (latestInTheQueue) {
+                    processStatusChangeEvent(event);
+                }
             }
         }
     }
@@ -345,7 +351,7 @@ private void processPendingHealthStatusChanges() {
     private void processStatusChangeEvent(HealthStatusChangeEvent eventArgs) {
         Endpoint endpoint = eventArgs.getEndpoint();
         HealthStatus newStatus = eventArgs.getNewStatus();
-        log.info("Health status changed for {} from {} to {}", endpoint, eventArgs.getOldStatus(), newStatus);
+        log.debug("Health status changed for {} from {} to {}", endpoint, eventArgs.getOldStatus(), newStatus);
 
         Cluster clusterWithHealthChange = multiClusterMap.get(endpoint);
 
@@ -394,7 +400,7 @@ private Cluster waitForInitialHealthyCluster() {
                 status = statusTracker.waitForHealthStatus(endpoint);
             } else {
                 // No health check configured - assume healthy
-                log.info("No health check configured for cluster {}, deafulting to HEALTHY", endpoint);
+                log.info("No health check configured for cluster {}, defaulting to HEALTHY", endpoint);
                 status = HealthStatus.HEALTHY;
             }