diff --git a/lib-jedis-pool/README.md b/lib-jedis-pool/README.md index 989bb032..4d1ab305 100644 --- a/lib-jedis-pool/README.md +++ b/lib-jedis-pool/README.md @@ -8,7 +8,7 @@ Add this dependency to your `build.gradle`: ```gradle dependencies { - implementation 'io.seqera:lib-jedis-pool:1.3.0' + implementation 'io.seqera:lib-jedis-pool:1.3.1' } ``` @@ -62,7 +62,8 @@ the shard, replicas included. A standalone client that lands on a replica gets ` command, and a replica still answers `PING`. With `redis.mode: cluster-primary`: - the host in `redis.uri` is used only to discover the shard primary (`CLUSTER SHARDS`, each candidate - confirmed with a probe); + confirmed with a probe). Only shards that own slots count: around a failover the reply can list the + failed primary as an extra shard with no slots (MemoryDB does, while it recovers the node); - pooled connections are validated on every borrow (`testOnBorrow` is forced on) by reading a probe key, `jedis-pool:primary-probe`: the primary answers, a replica answers `MOVED`. After a failover stale connections are evicted and new ones go to the new primary; @@ -70,7 +71,8 @@ command, and a replica still answers `PING`. With `redis.mode: cluster-primary`: Constraints: a single shard only, database 0 only, and no cross-slot multi-key commands. The Redis user needs `@slow` (for `CLUSTER SHARDS`) and read access to the probe key; `@admin`/`@dangerous` are not needed. -Commands in flight during a failover (a few seconds) fail and must be retried by the caller. +Commands issued during a failover fail until the server has promoted the replica (40 s in a forced +MemoryDB failover) and must be retried by the caller. ```yaml redis: diff --git a/lib-jedis-pool/VERSION b/lib-jedis-pool/VERSION index f0bb29e7..3a3cd8cc 100644 --- a/lib-jedis-pool/VERSION +++ b/lib-jedis-pool/VERSION @@ -1 +1 @@ -1.3.0 +1.3.1 diff --git a/lib-jedis-pool/changelog.txt b/lib-jedis-pool/changelog.txt index f5893030..bcdb1052 100644 --- a/lib-jedis-pool/changelog.txt +++ b/lib-jedis-pool/changelog.txt @@ -1,5 +1,13 @@ # lib-jedis-pool changelog +1.3.1 - 14 Oct 2026 +- Fix redis.mode=cluster-primary staying disconnected after a MemoryDB failover until the old primary + was recovered (minutes): a CLUSTER SHARDS reply listing the failed primary as an extra shard with no + slots was rejected as multi-shard. Only shards that own slots now count; the primaries of the others + remain candidates for the probe +- Include a summary of the CLUSTER SHARDS reply in discovery errors, and in the discovery log line when + the reply lists more than one shard + 1.3.0 - 7 Oct 2026 - Add redis.mode=cluster-primary for single-shard cluster-mode servers with replicas (e.g. AWS MemoryDB): connections go to the shard primary found via CLUSTER SHARDS, are validated on borrow with a keyed diff --git a/lib-jedis-pool/src/main/java/io/seqera/jedis/JedisConnector.java b/lib-jedis-pool/src/main/java/io/seqera/jedis/JedisConnector.java index 9ad10264..d1ebe407 100644 --- a/lib-jedis-pool/src/main/java/io/seqera/jedis/JedisConnector.java +++ b/lib-jedis-pool/src/main/java/io/seqera/jedis/JedisConnector.java @@ -220,13 +220,13 @@ protected HostAndPort discoverPrimary() { JedisException last = null; for (HostAndPort candidate : seedAddresses()) { try { - final List primaries; + final List shards; try (Jedis jedis = new Jedis(candidate, clientConfig)) { - primaries = primariesOf(jedis.clusterShards(), clientConfig.isSsl()); + shards = jedis.clusterShards(); } - for (HostAndPort node : primaries) { + for (HostAndPort node : primariesOf(shards, clientConfig.isSsl())) { if (confirmPrimary(node)) { - log.info("Redis cluster primary is {} (discovered via {})", node, candidate); + log.info("Redis cluster primary is {} (discovered via {}{})", node, candidate, shards.size() > 1 ? "; reply " + describe(shards) : ""); return node; } rejected.add(node); @@ -274,25 +274,78 @@ private List seedAddresses() { /** * The online nodes a {@code CLUSTER SHARDS} reply lists as primary, for a single-shard cluster. - * Usually one; briefly two right after a failover. + * Usually one; briefly more around a failover, which the caller tells apart with a probe. + * + *

Only a shard that owns slots holds data, so the cluster is single-shard as long as at most one + * shard owns slots. Around a failover the reply can carry entries that own none: MemoryDB lists the + * failed primary as a shard of its own while it is being recovered, and Redis 7.2 can report the + * shard with no slots right after a takeover. Their primaries are still candidates, after those of + * the shard that owns the slots. * * @param shards the reply * @param ssl whether connections use TLS, which selects {@code tls-port} when the node reports one - * @return the primaries' addresses - * @throws JedisException when the cluster has more than one shard, or no online primary + * @return the primaries' addresses, those of the shard that owns the slots first + * @throws JedisException when more than one shard owns slots, or no shard has an online primary */ static List primariesOf(List shards, boolean ssl) { - if (shards == null || shards.size() != 1) { - throw new JedisException("redis.mode=cluster-primary requires a single-shard cluster - found " + (shards == null ? 0 : shards.size()) + " shards"); + final List reply = shards != null ? shards : List.of(); + final List owning = reply.stream().filter(JedisConnector::ownsSlots).toList(); + if (owning.size() > 1) { + throw new JedisException("redis.mode=cluster-primary requires a single-shard cluster - found " + owning.size() + " shards owning slots: " + describe(reply)); + } + final List result = new ArrayList<>(2); + owning.forEach(shard -> addOnlinePrimaries(shard, ssl, result)); + reply.stream().filter(shard -> !ownsSlots(shard)).forEach(shard -> addOnlinePrimaries(shard, ssl, result)); + if (result.isEmpty()) { + throw new JedisConnectionException("No online primary in the Redis cluster: " + describe(reply)); } - final List result = new ArrayList<>(1); - for (ClusterShardNodeInfo node : shards.getFirst().getNodes()) { - if ("master".equals(node.getRole()) && "online".equals(node.getHealth())) { - result.add(new HostAndPort(hostOf(node), portOf(node, ssl))); + return result; + } + + private static boolean ownsSlots(ClusterShardInfo shard) { + return shard.getSlots() != null && !shard.getSlots().isEmpty(); + } + + private static void addOnlinePrimaries(ClusterShardInfo shard, boolean ssl, List result) { + for (ClusterShardNodeInfo node : shard.getNodes()) { + if (!"master".equals(node.getRole()) || !"online".equals(node.getHealth())) { + continue; + } + final HostAndPort address = new HostAndPort(hostOf(node), portOf(node, ssl)); + if (!result.contains(address)) { + result.add(address); } } - if (result.isEmpty()) { - throw new JedisConnectionException("No online primary in the Redis cluster shard"); + } + + /** + * A one-line summary of a {@code CLUSTER SHARDS} reply, e.g. + * {@code [slots=[0-16383] nodes=[10.0.0.2:6379 master online]; slots=[] nodes=[10.0.0.1:6379 master fail]]}, + * so a rejected reply shows what the server returned. + */ + static String describe(List shards) { + final StringBuilder result = new StringBuilder("["); + for (ClusterShardInfo shard : shards) { + if (result.length() > 1) { + result.append("; "); + } + result.append("slots=").append(slotsOf(shard)).append(" nodes=").append(nodesOf(shard)); + } + return result.append(']').toString(); + } + + private static List slotsOf(ClusterShardInfo shard) { + final List result = new ArrayList<>(); + for (List range : shard.getSlots() != null ? shard.getSlots() : List.>of()) { + result.add(range.size() == 2 ? range.get(0) + "-" + range.get(1) : range.toString()); + } + return result; + } + + private static List nodesOf(ClusterShardInfo shard) { + final List result = new ArrayList<>(); + for (ClusterShardNodeInfo node : shard.getNodes()) { + result.add(hostOf(node) + ":" + node.getPort() + " " + node.getRole() + " " + node.getHealth()); } return result; } diff --git a/lib-jedis-pool/src/test/groovy/io/seqera/jedis/JedisConnectorTest.groovy b/lib-jedis-pool/src/test/groovy/io/seqera/jedis/JedisConnectorTest.groovy index 21493d3c..df502dd9 100644 --- a/lib-jedis-pool/src/test/groovy/io/seqera/jedis/JedisConnectorTest.groovy +++ b/lib-jedis-pool/src/test/groovy/io/seqera/jedis/JedisConnectorTest.groovy @@ -30,6 +30,7 @@ import redis.clients.jedis.HostAndPortMapper import redis.clients.jedis.Jedis import redis.clients.jedis.JedisClientConfig import redis.clients.jedis.JedisPool +import redis.clients.jedis.exceptions.JedisConnectionException import redis.clients.jedis.exceptions.JedisException import redis.clients.jedis.resps.ClusterShardInfo import redis.clients.jedis.resps.ClusterShardNodeInfo @@ -175,6 +176,50 @@ class JedisConnectorTest extends Specification { thrown(JedisException) } + def 'should refuse a cluster whose slots are split across shards, and report the reply'() { + when: + JedisConnector.primariesOf([ + shard([[0L, 8191L]], node('master', '10.0.0.1', 6379, null)), + shard([[8192L, 16383L]], node('master', '10.0.0.2', 6379, null))], false) + + then: + def e = thrown(JedisException) + e.message.contains('found 2 shards owning slots') + e.message.contains('slots=[0-8191] nodes=[10.0.0.1:6379 master online]') + } + + def 'should see through the extra shard MemoryDB lists for a failed primary during recovery'() { + given: 'the promoted primary owns every slot, the old one is listed as a shard of its own with none' + def reply = [ + shard([], node('master', '10.0.0.1', 6379, null, OLD_HEALTH)), + shard([[0L, 16383L]], node('master', '10.0.0.2', 6379, null))] + + expect: 'the promoted primary comes first; the old one stays a candidate only while reported online' + JedisConnector.primariesOf(reply, false) == EXPECTED.collect { new HostAndPort(it, 6379) } + + where: + OLD_HEALTH | EXPECTED + 'fail' | ['10.0.0.2'] + 'online' | ['10.0.0.2', '10.0.0.1'] + } + + def 'should keep the primaries of a shard reported with no slots right after a takeover'() { + given: 'Redis 7.2, the replica having taken over from an unreachable primary' + def reply = [shard([], node('master', '10.0.0.1', 6379, null), node('master', '10.0.0.2', 6379, null))] + + expect: 'both are candidates, for the probe to tell apart' + JedisConnector.primariesOf(reply, false) == [new HostAndPort('10.0.0.1', 6379), new HostAndPort('10.0.0.2', 6379)] + } + + def 'should report the reply when no shard has an online primary'() { + when: + JedisConnector.primariesOf([shard([[0L, 16383L]], node('master', '10.0.0.1', 6379, null, 'fail'), node('replica', '10.0.0.2', 6379, null))], false) + + then: + def e = thrown(JedisConnectionException) + e.message == 'No online primary in the Redis cluster: [slots=[0-16383] nodes=[10.0.0.1:6379 master fail, 10.0.0.2:6379 replica online]]' + } + // -- helpers private GenericContainer clusterNode() { @@ -210,12 +255,16 @@ class JedisConnectorTest extends Specification { } private static ClusterShardInfo shard(ClusterShardNodeInfo... nodes) { - new ClusterShardInfo([(ClusterShardInfo.SLOTS): [[0L, 16383L]], (ClusterShardInfo.NODES): nodes as List]) + shard([[0L, 16383L]], nodes) + } + + private static ClusterShardInfo shard(List> slots, ClusterShardNodeInfo... nodes) { + new ClusterShardInfo([(ClusterShardInfo.SLOTS): slots, (ClusterShardInfo.NODES): nodes as List]) } - private static ClusterShardNodeInfo node(String role, String ip, long port, Long tlsPort) { + private static ClusterShardNodeInfo node(String role, String ip, long port, Long tlsPort, String health = 'online') { final map = [(ClusterShardNodeInfo.ID): ip, (ClusterShardNodeInfo.ENDPOINT): ip, (ClusterShardNodeInfo.IP): ip, - (ClusterShardNodeInfo.PORT): port, (ClusterShardNodeInfo.ROLE): role, (ClusterShardNodeInfo.HEALTH): 'online'] + (ClusterShardNodeInfo.PORT): port, (ClusterShardNodeInfo.ROLE): role, (ClusterShardNodeInfo.HEALTH): health] if (tlsPort != null) { map[ClusterShardNodeInfo.TLS_PORT] = tlsPort }