From 3f1085e6b8afda54214c63a0e27ec54d245d0ff2 Mon Sep 17 00:00:00 2001 From: HwangRock Date: Sat, 18 Jul 2026 09:36:09 +0900 Subject: [PATCH] Signal onError instead of hanging when a reactive complex-output command errors Reactive commands backed by EncodedComplexOutput (CF.INFO, BF.INFO, TOPK.LIST, TS.INFO, ...) hang until the command timeout fires when the server replies with an error. SubscriptionCommand.doOnComplete() calls getOutput().get() before checking hasError(); get() runs the ComplexDataParser, which throws on the null payload left by an error reply. That throw escapes doOnComplete() after the completion state has already flipped to COMPLETE, so completeExceptionally() skips onError and the Flux/Mono never terminates. Check hasError() before get(), matching AsyncCommand.completeResult() and MultiOutput. The normal path is unchanged; only error replies are affected, and they now surface as onError instead of hanging. --- src/main/java/io/lettuce/core/RedisPublisher.java | 4 ++-- .../RedisCuckooFilterReactiveIntegrationTests.java | 14 ++++++++++++++ 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/src/main/java/io/lettuce/core/RedisPublisher.java b/src/main/java/io/lettuce/core/RedisPublisher.java index 13797dc7bf..7bda884a89 100644 --- a/src/main/java/io/lettuce/core/RedisPublisher.java +++ b/src/main/java/io/lettuce/core/RedisPublisher.java @@ -761,13 +761,13 @@ protected void doOnComplete() { if (getOutput() != null) { - Object result = getOutput().get(); - if (getOutput().hasError()) { onError(ExceptionFactory.createExecutionException(getOutput().getError())); return; } + Object result = getOutput().get(); + if (!(getOutput() instanceof StreamingOutput) && result != null) { if (dissolve && result instanceof Collection) { diff --git a/src/test/java/io/lettuce/core/probabilistic/RedisCuckooFilterReactiveIntegrationTests.java b/src/test/java/io/lettuce/core/probabilistic/RedisCuckooFilterReactiveIntegrationTests.java index 799806433a..d33847e267 100644 --- a/src/test/java/io/lettuce/core/probabilistic/RedisCuckooFilterReactiveIntegrationTests.java +++ b/src/test/java/io/lettuce/core/probabilistic/RedisCuckooFilterReactiveIntegrationTests.java @@ -7,8 +7,10 @@ package io.lettuce.core.probabilistic; import javax.inject.Inject; +import java.time.Duration; import java.util.List; +import io.lettuce.core.RedisCommandExecutionException; import io.lettuce.core.Value; import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.api.reactive.RedisReactiveCommands; @@ -138,4 +140,16 @@ void cfInsertReturnsFalseWhenFilterIsFull() { } } + /** + * Reproduces the reactive hang on server errors: {@code RedisPublisher.SubscriptionCommand#doOnComplete()} used to call + * {@code getOutput().get()} before checking {@code getOutput().hasError()}, so a {@link CfInfoValueParser#parse} failure on + * a {@code null} payload (server error, no data) threw before the error signal could reach the subscriber, leaving the + * {@code Mono} hanging forever instead of erroring. + */ + @Test + void cfInfoOnMissingKeyErrorsInsteadOfHanging() { + StepVerifier.create(reactive.cfInfo("does-not-exist-key")).expectError(RedisCommandExecutionException.class) + .verify(Duration.ofSeconds(5)); + } + }