failoverListeners = new ArrayList<>();
+
+ // -------------------------------------------------------------------------
+ // Inner classes
+ // -------------------------------------------------------------------------
+
class TopologyRefreshTask implements Runnable {
@Override
public void run() {
@@ -62,6 +99,94 @@ public void run() {
}
}
+ /**
+ * Background thread that subscribes to the {@value #CLUSTER_FAILOVER_CHANNEL} channel on
+ * a single cluster node. When a failover notification is received the thread immediately
+ * calls {@link #renewClusterSlots(Connection)} so that the slot cache is updated without
+ * waiting for the next periodic refresh cycle.
+ *
+ * The thread reconnects automatically after connection failures, using an exponential
+ * back-off capped at {@value #MAX_RETRY_WAIT_MILLIS} ms.
+ */
+ class FailoverListener extends Thread {
+
+ private static final long INITIAL_RETRY_WAIT_MILLIS = 1_000L;
+ private static final long MAX_RETRY_WAIT_MILLIS = 30_000L;
+
+ private final HostAndPort node;
+ private volatile Jedis jedis;
+ private volatile boolean running = false;
+
+ FailoverListener(HostAndPort node) {
+ super("ClusterFailoverListener-[" + node + "]");
+ this.node = node;
+ setDaemon(true);
+ }
+
+ @Override
+ public void run() {
+ running = true;
+ long retryWaitMillis = INITIAL_RETRY_WAIT_MILLIS;
+
+ while (running) {
+ try {
+ if (!running) break;
+
+ jedis = new Jedis(node, clientConfig);
+
+ // Perform an immediate refresh when (re-)connecting so that any failover
+ // that happened while the listener was disconnected is not missed.
+ logger.debug("FailoverListener connected to {}, performing initial slot refresh.", node);
+ renewClusterSlots(jedis.getClient());
+
+ // Reset back-off on successful connection.
+ retryWaitMillis = INITIAL_RETRY_WAIT_MILLIS;
+
+ jedis.subscribe(new JedisPubSub() {
+ @Override
+ public void onMessage(String channel, String message) {
+ logger.info(
+ "Cluster failover notification received from {} on channel '{}': {}",
+ node, channel, message);
+ // Trigger an immediate topology refresh. We pass null so that the
+ // implementation falls back to startNodes / shuffled pool for the
+ // CLUSTER SLOTS query, which is safer than reusing the subscriber
+ // connection that is currently blocked in the subscribe loop.
+ renewClusterSlots(null);
+ }
+ }, CLUSTER_FAILOVER_CHANNEL);
+
+ } catch (JedisException e) {
+ if (running) {
+ logger.warn(
+ "FailoverListener lost connection to {}. Retrying in {}ms.", node, retryWaitMillis, e);
+ try {
+ Thread.sleep(retryWaitMillis);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ logger.debug("FailoverListener interrupted while sleeping.", ie);
+ }
+ // Exponential back-off.
+ retryWaitMillis = Math.min(retryWaitMillis * 2, MAX_RETRY_WAIT_MILLIS);
+ } else {
+ logger.debug("FailoverListener shutting down for {}.", node);
+ }
+ } finally {
+ IOUtils.closeQuietly(jedis);
+ }
+ }
+ }
+
+ void shutdown() {
+ running = false;
+ IOUtils.closeQuietly(jedis);
+ }
+ }
+
+ // -------------------------------------------------------------------------
+ // Constructors
+ // -------------------------------------------------------------------------
+
public JedisClusterInfoCache(final JedisClientConfig clientConfig, final Set startNodes) {
this(clientConfig, null, startNodes);
}
@@ -85,6 +210,49 @@ public JedisClusterInfoCache(final JedisClientConfig clientConfig,
}
}
+ // -------------------------------------------------------------------------
+ // Failover listener management
+ // -------------------------------------------------------------------------
+
+ /**
+ * Enable or disable the failover broadcast listener. Must be called before the cache is
+ * used (i.e. before {@link #discoverClusterNodesAndSlots(Connection)} is invoked).
+ *
+ * @param enabled {@code true} to enable (default), {@code false} to disable
+ */
+ public void setFailoverListenerEnabled(boolean enabled) {
+ this.failoverListenerEnabled = enabled;
+ }
+
+ /**
+ * Start a {@link FailoverListener} for every node currently known to the cache.
+ * Called automatically after the initial slot discovery.
+ */
+ private void startFailoverListeners() {
+ if (!failoverListenerEnabled || startNodes == null || startNodes.isEmpty()) {
+ return;
+ }
+ // Stop any existing listeners first (e.g. after a full reset).
+ stopFailoverListeners();
+ for (HostAndPort node : startNodes) {
+ FailoverListener listener = new FailoverListener(node);
+ failoverListeners.add(listener);
+ listener.start();
+ logger.info("Started FailoverListener for cluster node {}.", node);
+ }
+ }
+
+ private void stopFailoverListeners() {
+ for (FailoverListener listener : failoverListeners) {
+ listener.shutdown();
+ }
+ failoverListeners.clear();
+ }
+
+ // -------------------------------------------------------------------------
+ // Slot discovery
+ // -------------------------------------------------------------------------
+
/**
* Check whether the number and order of slots in the cluster topology are equal to CLUSTER_HASHSLOTS
* @param slotsInfo the cluster topology
@@ -148,6 +316,8 @@ public void discoverClusterNodesAndSlots(Connection jedis) {
} finally {
w.unlock();
}
+ // Start failover listeners after the initial topology is known.
+ startFailoverListeners();
}
public void renewClusterSlots(Connection jedis) {
@@ -381,6 +551,7 @@ public void reset() {
public void close() {
reset();
+ stopFailoverListeners();
if (topologyRefreshExecutor != null) {
logger.info("Cluster topology refresh shutdown, startNodes: {}", startNodes);
topologyRefreshExecutor.shutdownNow();
diff --git a/src/main/java/io/valkey/providers/SentineledConnectionProvider.java b/src/main/java/io/valkey/providers/SentineledConnectionProvider.java
index 857c2bad..f316f48e 100644
--- a/src/main/java/io/valkey/providers/SentineledConnectionProvider.java
+++ b/src/main/java/io/valkey/providers/SentineledConnectionProvider.java
@@ -4,6 +4,9 @@
import java.util.Collection;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import io.valkey.CommandArguments;
@@ -21,12 +24,38 @@
import io.valkey.exceptions.JedisConnectionException;
import io.valkey.exceptions.JedisException;
+/**
+ * Sentinel-mode connection provider that combines two complementary failover-detection
+ * strategies:
+ *
+ *
+ * - Pub/Sub broadcast listener (existing): each {@link SentinelListener} thread
+ * subscribes to the {@code +switch-master} channel on its sentinel node and calls
+ * {@link #initMaster} immediately when a failover event is published.
+ * - Periodic active probe (new, feature/failover-enhancement): a separate
+ * single-threaded scheduler calls {@code SENTINEL GETMASTERADDRBYNAME} on every
+ * sentinel at a configurable interval. This ensures that failovers are detected even
+ * when the Pub/Sub connection is temporarily disrupted or when the sentinel does not
+ * publish a notification (e.g. due to a network partition).
+ *
+ *
+ * The probe interval defaults to {@value #DEFAULT_PROBE_PERIOD_MILLIS} ms and can be
+ * customised via the constructor that accepts {@code probePeriodMillis}. Set the interval
+ * to {@code 0} or a negative value to disable the active probe entirely.
+ */
public class SentineledConnectionProvider implements ConnectionProvider {
private static final Logger LOG = LoggerFactory.getLogger(SentineledConnectionProvider.class);
protected static final long DEFAULT_SUBSCRIBE_RETRY_WAIT_TIME_MILLIS = 5000;
+ /**
+ * Default interval (ms) for the periodic active-probe task.
+ * 10 seconds is a reasonable default: fast enough to catch missed events,
+ * slow enough not to flood sentinels with queries.
+ */
+ public static final long DEFAULT_PROBE_PERIOD_MILLIS = 10_000L;
+
private volatile HostAndPort currentMaster;
private volatile ConnectionPool pool;
@@ -43,8 +72,27 @@ public class SentineledConnectionProvider implements ConnectionProvider {
private final long subscribeRetryWaitTimeMillis;
+ /**
+ * Interval (ms) for the active-probe scheduler. {@code <= 0} disables the probe.
+ */
+ private final long probePeriodMillis;
+
+ /**
+ * The sentinel nodes used for active probing (same set as the listeners).
+ */
+ private volatile Set sentinelNodes;
+
+ /**
+ * Single-threaded scheduler that periodically queries sentinels for the current master.
+ */
+ private ScheduledExecutorService probeExecutor = null;
+
private final Object initPoolLock = new Object();
+ // -------------------------------------------------------------------------
+ // Constructors
+ // -------------------------------------------------------------------------
+
public SentineledConnectionProvider(String masterName, final JedisClientConfig masterClientConfig,
Set sentinels, final JedisClientConfig sentinelClientConfig) {
this(masterName, masterClientConfig, /*poolConfig*/ null, sentinels, sentinelClientConfig);
@@ -61,6 +109,20 @@ public SentineledConnectionProvider(String masterName, final JedisClientConfig m
final GenericObjectPoolConfig poolConfig,
Set sentinels, final JedisClientConfig sentinelClientConfig,
final long subscribeRetryWaitTimeMillis) {
+ this(masterName, masterClientConfig, poolConfig, sentinels, sentinelClientConfig,
+ subscribeRetryWaitTimeMillis, DEFAULT_PROBE_PERIOD_MILLIS);
+ }
+
+ /**
+ * Full constructor that exposes the active-probe interval.
+ *
+ * @param probePeriodMillis interval between active-probe cycles in milliseconds;
+ * {@code <= 0} disables the active probe
+ */
+ public SentineledConnectionProvider(String masterName, final JedisClientConfig masterClientConfig,
+ final GenericObjectPoolConfig poolConfig,
+ Set sentinels, final JedisClientConfig sentinelClientConfig,
+ final long subscribeRetryWaitTimeMillis, final long probePeriodMillis) {
this.masterName = masterName;
this.masterClientConfig = masterClientConfig;
@@ -68,11 +130,20 @@ public SentineledConnectionProvider(String masterName, final JedisClientConfig m
this.sentinelClientConfig = sentinelClientConfig;
this.subscribeRetryWaitTimeMillis = subscribeRetryWaitTimeMillis;
+ this.probePeriodMillis = probePeriodMillis;
HostAndPort master = initSentinels(sentinels);
initMaster(master);
+
+ if (probePeriodMillis > 0) {
+ startActiveProbe(sentinels);
+ }
}
+ // -------------------------------------------------------------------------
+ // ConnectionProvider interface
+ // -------------------------------------------------------------------------
+
@Override
public Connection getConnection() {
return pool.getResource();
@@ -86,7 +157,7 @@ public Connection getConnection(CommandArguments args) {
@Override
public void close() {
sentinelListeners.forEach(SentinelListener::shutdown);
-
+ stopActiveProbe();
pool.close();
}
@@ -94,6 +165,10 @@ public HostAndPort getCurrentMaster() {
return currentMaster;
}
+ // -------------------------------------------------------------------------
+ // Master pool management
+ // -------------------------------------------------------------------------
+
private void initMaster(HostAndPort master) {
synchronized (initPoolLock) {
if (!master.equals(currentMaster)) {
@@ -117,6 +192,10 @@ private void initMaster(HostAndPort master) {
}
}
+ // -------------------------------------------------------------------------
+ // Sentinel initialisation
+ // -------------------------------------------------------------------------
+
private HostAndPort initSentinels(Set sentinels) {
HostAndPort master = null;
@@ -175,6 +254,84 @@ private HostAndPort initSentinels(Set sentinels) {
return master;
}
+ // -------------------------------------------------------------------------
+ // Active probe scheduler
+ // -------------------------------------------------------------------------
+
+ /**
+ * Start the periodic active-probe scheduler.
+ *
+ * The probe task iterates over all sentinel nodes and calls
+ * {@code SENTINEL GETMASTERADDRBYNAME} on the first reachable one. If the returned
+ * master address differs from {@link #currentMaster} the pool is updated immediately,
+ * providing a safety net for cases where the Pub/Sub notification was missed.
+ *
+ * @param sentinels the set of sentinel nodes to probe
+ */
+ private void startActiveProbe(Set sentinels) {
+ this.sentinelNodes = sentinels;
+ probeExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread t = new Thread(r, "SentinelActiveProbe-" + masterName);
+ t.setDaemon(true);
+ return t;
+ });
+ probeExecutor.scheduleWithFixedDelay(
+ new ActiveProbeTask(),
+ probePeriodMillis,
+ probePeriodMillis,
+ TimeUnit.MILLISECONDS);
+ LOG.info("Sentinel active probe started for master '{}', interval={}ms.", masterName, probePeriodMillis);
+ }
+
+ private void stopActiveProbe() {
+ if (probeExecutor != null) {
+ probeExecutor.shutdownNow();
+ probeExecutor = null;
+ LOG.info("Sentinel active probe stopped for master '{}'.", masterName);
+ }
+ }
+
+ /**
+ * Runnable executed by the active-probe scheduler.
+ * Queries each sentinel in turn until one responds, then updates the master pool if needed.
+ */
+ private class ActiveProbeTask implements Runnable {
+
+ @Override
+ public void run() {
+ if (sentinelNodes == null) return;
+
+ for (HostAndPort sentinel : sentinelNodes) {
+ try (Jedis jedis = new Jedis(sentinel, sentinelClientConfig)) {
+ List masterAddr = jedis.sentinelGetMasterAddrByName(masterName);
+ if (masterAddr != null && masterAddr.size() == 2) {
+ HostAndPort probedMaster = toHostAndPort(masterAddr);
+ if (!probedMaster.equals(currentMaster)) {
+ LOG.info(
+ "Active probe detected master change for '{}': {} -> {}. Updating pool.",
+ masterName, currentMaster, probedMaster);
+ initMaster(probedMaster);
+ } else {
+ LOG.debug("Active probe: master '{}' still at {}.", masterName, currentMaster);
+ }
+ // Successfully queried one sentinel – no need to try the rest.
+ return;
+ } else {
+ LOG.warn("Active probe: sentinel {} returned no address for master '{}'.", sentinel, masterName);
+ }
+ } catch (JedisException e) {
+ LOG.warn("Active probe: could not reach sentinel {}. Trying next.", sentinel, e);
+ }
+ }
+
+ LOG.warn("Active probe: all sentinels unreachable for master '{}'.", masterName);
+ }
+ }
+
+ // -------------------------------------------------------------------------
+ // Helpers
+ // -------------------------------------------------------------------------
+
/**
* Must be of size 2.
*/
@@ -186,6 +343,18 @@ private static HostAndPort toHostAndPort(String hostStr, String portStr) {
return new HostAndPort(hostStr, Integer.parseInt(portStr));
}
+ // -------------------------------------------------------------------------
+ // SentinelListener inner class
+ // -------------------------------------------------------------------------
+
+ /**
+ * Background thread that subscribes to the {@code +switch-master} Pub/Sub channel on a
+ * single sentinel node. When a failover event is published the thread immediately calls
+ * {@link #initMaster} to switch the connection pool to the new master.
+ *
+ * The thread reconnects automatically after connection failures, sleeping
+ * {@link #subscribeRetryWaitTimeMillis} ms between attempts.
+ */
protected class SentinelListener extends Thread {
protected final HostAndPort node;
@@ -212,7 +381,8 @@ public void run() {
sentinelJedis = new Jedis(node, sentinelClientConfig);
- // code for active refresh
+ // Perform an active refresh immediately on (re-)connect so that any failover
+ // that happened while the listener was disconnected is not missed.
List masterAddr = sentinelJedis.sentinelGetMasterAddrByName(masterName);
if (masterAddr == null || masterAddr.size() != 2) {
LOG.warn("Cannot get master {} address. Sentinel: {}.", masterName, node);