StreamMessageListenerContainer stops consuming messages after connection timeout to Redis

[radjavi]

Problem

I have been following this guide for implementing asynchronous message listeners for Redis streams. The listeners work as expected at first. However, if a temporary connection timeout occurs towards Redis (e.g. Redis becomes unavailable), the listeners stop consuming messages indefinitely (even though Redis is available).

How to reproduce

1. Create a message listener using this guide (with a StreamMessageListenerContainer).
2. Publish an event (to make sure the message is successfully received).
3. Restart the Redis instance (to simulate a connection timeout)
4. Publish an event (should not be received)

[christophstrobl]

Thank you for getting in touch. Which Redis driver are you using? Lettuce has a connection watchdog that can trigger reconnect automatically depending on the options set.

If you'd like us to spend some time investigating, please take the time to provide a complete minimal sample (something that we can unzip or git clone, build, and deploy) that reproduces the problem.

[radjavi]

I have tried both Lettuce and Jedis. Same issue with both. Worth mentioning is that code in other places (e.g. an event publisher) that also connects to Redis works fine after Redis becomes available again. It's only the message listeners that stop working.

I'll see if I get some time over to create an example project.

[zhanglc]

seems like the spring-data-redis auto unsubscribe when exception.

spring-data-redis/src/main/java/org/springframework/data/redis/stream/StreamPollTask.java

Lines 136 to 138 in a7d3914

	if (cancelSubscriptionOnError.test(ex)) {
	cancel();
	}

the value of cancelSubscriptionOnError default is true:

spring-data-redis/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java

Line 337 in a7d3914

Predicate<Throwable> cancelSubscriptionOnError = t -> true;

when the RuntimeException occurs, spring does not consume the message any more, so, we just need change this options in builder:

        Subscription subscription =
                listenerContainer.register(
                        StreamMessageListenerContainer.StreamReadRequest
                                .builder(StreamOffset
                                        .create(REDIS_STREAM_KEY_REFRESH_RELATION, ReadOffset.lastConsumed()))
                                // here
                                .cancelOnError(t -> false)
                                .consumer(Consumer.from(REDIS_STREAM_KEY_REFRESH_RELATION_GROUP_NAME,
                                        REDIS_STREAM_KEY_REFRESH_RELATION_CONSUMER_NAME))
                                .autoAcknowledge(true).build(),
                        streamListener);

by the way ,i think the default value of cancelSubscriptionOnError should be t -> false .