Signal onError instead of hanging when a reactive complex-output command errors

[HwangRock]

Fixes #3855

Motivation

Reactive commands backed by EncodedComplexOutput — CF.INFO, BF.INFO, TOPK.INFO, TOPK.LIST, and any other command whose output is decoded by a ComplexDataParser — hang until the command timeout fires when the server replies with an error.

Reproduction against a live server:

StepVerifier.create(reactive.cfInfo("does-not-exist-key"))
    .expectError(RedisCommandExecutionException.class)
    .verify(Duration.ofSeconds(5));

Before this change the Mono never terminates; the subscriber waits until the client-side command timeout (default 60s).

Root cause

SubscriptionCommand.doOnComplete() in RedisPublisher calls getOutput().get() before checking hasError():

Object result = getOutput().get();      // runs the parser
if (getOutput().hasError()) { ... }     // never reached

On an error reply the output holds an error and a null payload. get() runs the ComplexDataParser, which throws IllegalArgumentException on the null data. That throw escapes doOnComplete(), which runs inside CommandWrapper.complete() after the completion state has already been flipped to COMPLETE. CommandHandler catches the throw and calls completeExceptionally(), but its consumers != COMPLETE guard is now false, so onError is never signalled and the Flux/Mono never terminates.

The non-reactive paths do not have this problem because they check hasError() first: AsyncCommand.completeResult() and MultiOutput both surface the error without ever calling get(). Only the reactive path had the ordering reversed. This was latent until ComplexDataParser-based outputs (RESP3 complex replies) made get() able to throw.

Modification

Check hasError() before get() in doOnComplete(), matching AsyncCommand.completeResult():

if (getOutput().hasError()) {
    onError(ExceptionFactory.createExecutionException(getOutput().getError()));
    return;
}
Object result = getOutput().get();

The dissolve/onNext/StreamingOutput logic is unchanged. The normal path (hasError() == false) is byte-for-byte identical; only error replies change behaviour, and they now surface as onError instead of hanging. Normal empty replies (hasError() == false, data == null) are unaffected — the parser returns null as before.

Result

• Added cfInfoOnMissingKeyErrorsInsteadOfHanging to RedisCuckooFilterReactiveIntegrationTests. Fails with a 5s timeout before the change, passes after.
• Existing reactive happy-path integration tests for CF/BF/TopK pass unchanged (50 tests): the normal path is untouched.
• Verified against redis:8.

[a-TODO-rov]

Hey @HwangRock
Could you please open an issue (or a Bug) for that one first. Seems that a lot of commands would be affected.

[HwangRock]

@a-TODO-rov Done, opened #3855 with the repro, root cause, and affected scope. This PR is the fix for it.