RReliableTopic: listener silently never consumes — `poll()`

Redis version

MemoryDB 7.1

Redisson version

4.6.1

Redisson configuration

ClusterServersConfig, readMode left at its default (ReadMode.SLAVE)

What is the Expected behavior?

Suggested fixes

1. Route poll()'s pending read to the master. It is the head of a consumer read-modify-write loop against state this client just wrote, so a replica cannot answer it correctly. Either add a master-pinned pendingRange variant used only from poll(), or issue the script via evalWriteAsync there.
2. Do not treat NOGROUP as terminal without confirming it against the master, and log before giving up regardless. A bounded retry via the existing newTimeout path would make the replica-lag case self-healing.
3. If the group truly is gone, clean up rather than leaving a phantom subscriber. Remove the timeout-ZSET entry, clear subscribed, and surface the condition to the caller (exception on the addListenerAsync future, or a listener callback) so it can resubscribe. As it stands, publishAsync() reports successful delivery to a subscriber that cannot receive.

Fix 1 alone resolves the replica race. Fix 2 and 3 matter independently: the same silent death occurs on a single node whenever the group legitimately disappears while a listener is live — for example if the topic key is given a TTL that expires — and the caller has no way to detect it.

Workaround

Set ReadMode.MASTER on the cluster config. This removes the replica read entirely and the failure disappears. It does not address problems 2 and 3.

What is the Actual behavior?

Summary

With RReliableTopic on a Redis Cluster using the default ReadMode.SLAVE, a large fraction of addListener() calls produce a listener that never issues XREADGROUP and never receives a message, permanently, with nothing logged.

In production we measured 22 of 37 subscribers (59%) dead this way over a 44-minute MONITOR capture on one master.

The failure is silent in both directions, which is what makes it severe:

• the subscriber never reads, and no error is logged or surfaced to the caller;
• the consumer group and the timeout-ZSET entry are left in place, so publishAsync() counts the dead subscriber via xinfo groups and returns a non-zero delivery count. The publisher receives positive acknowledgement for a message that will never be read.

Root cause

Two independent problems compound. Either alone would be recoverable.

1. poll() reads consumer-group state from a replica immediately after writing it to the master

addListenerAsync() (RedissonReliableTopic.java:145-166) creates the group on the master:

RFuture<Void> addFuture = commandExecutor.evalWriteNoRetryAsync(getRawName(), StringCodec.INSTANCE, RedisCommands.EVAL_VOID,
                  "redis.call('zadd', KEYS[2], ARGV[3], ARGV[2]);" +
                        "redis.call('xgroup', 'create', KEYS[1], ARGV[2], ARGV[1], 'MKSTREAM'); ",
        ...

CompletionStage<String> f = addFuture.thenApply(r -> {
    renewExpiration();
    poll(subscriberId);
    return id;
});

poll() (RedissonReliableTopic.java:169-170) then opens the consumer loop with a pendingRange call, and only reaches the blocking XREADGROUP from that future's thenCompose:

private void poll(String id) {
    RFuture<Map<StreamMessageId, Map<String, Object>>> f = stream.pendingRangeAsync(id, StreamMessageId.MIN, StreamMessageId.MAX, 100);
    CompletionStage<...> ff = f.thenCompose(r -> {
        ...
        if (r.isEmpty()) {
            readFuture = stream.readGroupAsync(id, "consumer",
                                    StreamReadGroupArgs.neverDelivered().timeout(Duration.ofSeconds(0)));
            return readFuture;
        }

RedissonStream.pendingRangeAsync (RedissonStream.java:833-845) uses evalReadAsync:

return commandExecutor.evalReadAsync(getRawName(), codec, EVAL_XRANGE,
        "local pendingData = redis.call('xpending', KEYS[1], ARGV[1], ARGV[2], ARGV[3], ARGV[4]);" +
        ...

Under ReadMode.SLAVE that script executes on a replica, roughly 4 ms after the XGROUP CREATE was applied on the master. Because the script carries no shebang it is not flagged as a write, so cluster redirection does not bounce it, and the replica (which has READONLY set by Redisson at connection setup) executes it locally against its own asynchronously-replicated copy of the stream. If the group has not replicated yet, redis.call('xpending', ...) raises:

NOGROUP No such key 'ws:...' or consumer group '<subscriberId>' in XPENDING

This is a read-after-write on state written microseconds earlier. There is no replication-lag threshold that makes it correct.

2. NOGROUP is treated as terminal — silently, with no cleanup

RedissonReliableTopic.java:186-196:

ff.whenComplete((res, ex) -> {
    if (ex != null) {
        if (getServiceManager().isShuttingDown(ex)) {
            return;
        }

        if (ex.getCause() != null
                && ex.getCause().getMessage().contains("NOGROUP")) {
            return;                                     // <-- no log, no retry, no cleanup
        }

        log.error("Unable to poll a new element. Subscriber id: {}", id, ex.getCause());

        getServiceManager().newTimeout(task -> { ... poll(id) ... }, 1, TimeUnit.SECONDS);

Every other error path logs and retries after 1s. NOGROUP alone returns.

This branch is presumably intended for the legitimate case where the group was destroyed by removeListenerAsync or the expired-subscriber sweep, where polling should stop. But NOGROUP from a lagging replica is indistinguishable from NOGROUP because the group is genuinely gone, and the code takes the interpretation that kills the listener permanently.

Nothing else recovers it:

• readFuture is never assigned, so no XREADGROUP is ever sent;
• subscribed stays true and the timeout-ZSET entry is left in place, so publishAsync() keeps counting the subscriber and countSubscribers() keeps reporting it;
• the watchdog (renewExpiration, RedissonReliableTopic.java:341-367, interval reliableTopicWatchdogTimeout / 3) keeps the registration alive until the ZSET entry disappears for unrelated reasons.

Evidence

MONITOR on one cluster master, 549 lines, 43m41s, two application nodes. Keys and subscriber ids abbreviated.

37 XGROUP CREATE, 15 XREADGROUP. Zero xpending on the master — a MONITOR on the replica for the same slot range shows the xpending calls landing there, confirming the routing.

Working subscriber — XREADGROUP follows 4 ms after the create (replica had caught up):

50.434  EVALSHA <add>    lua zadd  {ws:A}:timeout <ttl> 5cbc0ae2
50.434                   lua xgroup create ws:A 5cbc0ae2 0 MKSTREAM
50.438  XREADGROUP GROUP 5cbc0ae2 consumer BLOCK 0 STREAMS ws:A >
54.434  EVALSHA <publish> lua xadd ws:A * m {...}
54.434                    lua xinfo groups ws:A
54.442  XACK ws:A 5cbc0ae2 1785264015404-0            <-- delivered

Dead subscriber — no XREADGROUP, ever:

142.954  EVALSHA <add>     lua zadd  {ws:B}:timeout <ttl> 9e489503
142.954                    lua xgroup create ws:B 9e489503 0 MKSTREAM
                           (no XREADGROUP)
155.281  EVALSHA <publish> lua xadd ws:B * m {...}
155.281                    lua xinfo groups ws:B      <-- returns 1, publish() returns 1
343.022  EVALSHA <renew>   lua zscore {ws:B}:timeout 9e489503   <-- watchdog at create+200.0s
                           (silence, forever)

All 22 failures have exactly this shape. All 15 successes show XREADGROUP 4–10 ms after the create. The create + 200.0s watchdog tick matches the default reliableTopicWatchdogTimeout of 600000 ms divided by 3.

Reproduction

1. Redis Cluster with at least one replica per shard.
2. Redisson with ClusterServersConfig and default readMode (SLAVE).
3. Inflate the replication link latency past the create→poll gap (~4 ms) — tc qdisc add dev eth0 root netem delay 50ms on a replica, or a proxy such as toxiproxy. -- this was originally found on an AWS MemoryoDB cluster with 2 nodes in 2 availability zones, so larger replication latency
4. In a loop: create an RReliableTopic, addListener(...), then publish(...) after 1s.

Observed: on the master, XGROUP CREATE with no matching XREADGROUP; publish() returns ≥1; countSubscribers() returns ≥1; the listener never fires; nothing is logged.

Expected: the listener consumes the message, or the failure is surfaced.

Additional information

No response