From 573bd4b834a8180d2dc64b877a04b3e3f541a6a7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20Mart=C3=AD?= Date: Wed, 14 Oct 2026 15:39:53 +0200 Subject: [PATCH] fix(lib-jedis-pool): stay connected through a MemoryDB failover that lists the failed primary as an extra shard (1.3.1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit In cluster-primary mode, primary discovery rejected any CLUSTER SHARDS reply that was not exactly one shard. After a MemoryDB failover the promoted primary lists the failed one as a shard of its own, with no slots, until the node is recovered, so sched stayed disconnected for about 4 minutes after MemoryDB had finished failing over. Count only the shards that own slots: more than one is still refused as a multi-shard cluster. The primaries of every entry remain candidates, those of the shard that owns the slots first, and the existing probe tells the real primary apart. Keeping the slotless entries matters: Redis 7.2 reports the shard with no slots right after a takeover, and the failover integration test fails if they are dropped. Discovery errors now carry a summary of the reply, and the discovery log line includes it when the reply lists more than one shard. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jon Martí --- lib-jedis-pool/README.md | 8 +- lib-jedis-pool/VERSION | 2 +- lib-jedis-pool/changelog.txt | 8 ++ .../java/io/seqera/jedis/JedisConnector.java | 83 +++++++++++++++---- .../io/seqera/jedis/JedisConnectorTest.groovy | 55 +++++++++++- 5 files changed, 134 insertions(+), 22 deletions(-) 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 }