Redis command timed out; nested exception is io.lettuce.core.RedisCommandTimeoutException: Command timed out after 1 minute(s)

Question

Today when I using Java redis client(spring-data-redis-2.3.9.RELEASE.jar) to consume Redis stream message encount this error:

2021-05-11 17:49:42.134 ERROR 26301 --- [-post-service-1] c.d.s.p.common.mq.StreamConsumerRunner   : pull message error

org.springframework.dao.QueryTimeoutException: Redis command timed out; nested exception is io.lettuce.core.RedisCommandTimeoutException: Command timed out after 1 minute(s)
    at org.springframework.data.redis.connection.lettuce.LettuceExceptionConverter.convert(LettuceExceptionConverter.java:70) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.connection.lettuce.LettuceExceptionConverter.convert(LettuceExceptionConverter.java:41) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.PassThroughExceptionTranslationStrategy.translate(PassThroughExceptionTranslationStrategy.java:44) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.FallbackExceptionTranslationStrategy.translate(FallbackExceptionTranslationStrategy.java:42) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.connection.lettuce.LettuceConnection.convertLettuceAccessException(LettuceConnection.java:275) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.connection.lettuce.LettuceStreamCommands.convertLettuceAccessException(LettuceStreamCommands.java:712) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.connection.lettuce.LettuceStreamCommands.xReadGroup(LettuceStreamCommands.java:602) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.connection.DefaultedRedisConnection.xReadGroup(DefaultedRedisConnection.java:591) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.core.DefaultStreamOperations$4.inRedis(DefaultStreamOperations.java:310) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.core.DefaultStreamOperations$RecordDeserializingRedisCallback.doInRedis(DefaultStreamOperations.java:376) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.core.DefaultStreamOperations$RecordDeserializingRedisCallback.doInRedis(DefaultStreamOperations.java:371) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:228) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:188) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.core.AbstractOperations.execute(AbstractOperations.java:96) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.core.DefaultStreamOperations.read(DefaultStreamOperations.java:305) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.stream.DefaultStreamMessageListenerContainer.lambda$getReadFunction$3(DefaultStreamMessageListenerContainer.java:236) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.stream.StreamPollTask.doLoop(StreamPollTask.java:138) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at org.springframework.data.redis.stream.StreamPollTask.run(StreamPollTask.java:123) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    at misc.config.async.pool.MdcTaskDecorator.lambda$decorate$0(MdcTaskDecorator.java:29) ~[dolphin-common-1.0.0-SNAPSHOT.jar!/:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[na:na]
    at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
Caused by: io.lettuce.core.RedisCommandTimeoutException: Command timed out after 1 minute(s)
    at io.lettuce.core.internal.ExceptionFactory.createTimeoutException(ExceptionFactory.java:53) ~[lettuce-core-6.1.1.RELEASE.jar!/:6.1.1.RELEASE]
    at io.lettuce.core.internal.Futures.awaitOrCancel(Futures.java:246) ~[lettuce-core-6.1.1.RELEASE.jar!/:6.1.1.RELEASE]
    at io.lettuce.core.FutureSyncInvocationHandler.handleInvocation(FutureSyncInvocationHandler.java:75) ~[lettuce-core-6.1.1.RELEASE.jar!/:6.1.1.RELEASE]
    at io.lettuce.core.internal.AbstractInvocationHandler.invoke(AbstractInvocationHandler.java:80) ~[lettuce-core-6.1.1.RELEASE.jar!/:6.1.1.RELEASE]
    at com.sun.proxy.$Proxy149.xreadgroup(Unknown Source) ~[na:na]
    at org.springframework.data.redis.connection.lettuce.LettuceStreamCommands.xReadGroup(LettuceStreamCommands.java:600) ~[spring-data-redis-2.3.9.RELEASE.jar!/:2.3.9.RELEASE]
    ... 15 common frames omitted

For the beginning the message could consume by the client successfully, some message could be consumed and handle correctly, after a while it will encount the time out problem. this is my code:

package com.dolphin.soa.post.common.mq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.ErrorHandler;

import java.time.Duration;

/**
 * @author dolphin
 */
@Component
@Slf4j
public class StreamConsumerRunner implements ApplicationRunner, DisposableBean {

    @Value("${dolphin.redis.stream.consumer}")
    private String consumer;

    @Value("${dolphin.redis.stream.group}")
    private String groupName;

    private final RedisConnectionFactory redisConnectionFactory;

    private final ThreadPoolTaskExecutor threadPoolTaskExecutor;

    private final StreamMessageListener streamMessageListener;

    private final StringRedisTemplate stringRedisTemplate;

    private StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer;

    public StreamConsumerRunner(RedisConnectionFactory redisConnectionFactory,
                                ThreadPoolTaskExecutor threadPoolTaskExecutor,
                                StreamMessageListener streamMessageListener,
                                StringRedisTemplate stringRedisTemplate) {
        this.redisConnectionFactory = redisConnectionFactory;
        this.threadPoolTaskExecutor = threadPoolTaskExecutor;
        this.streamMessageListener = streamMessageListener;
        this.stringRedisTemplate = stringRedisTemplate;
    }

    @Override
    public void run(ApplicationArguments args) {
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> streamMessageListenerContainerOptions = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                .batchSize(1)
                .executor(this.threadPoolTaskExecutor)
                .errorHandler(new ErrorHandler() {
                    @Override
                    public void handleError(Throwable t) {
                        log.error("pull message error",t);
                    }
                })
                .pollTimeout(Duration.ofMinutes(1L))
                .serializer(new StringRedisSerializer())
                .build();
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer = StreamMessageListenerContainer
                .create(this.redisConnectionFactory, streamMessageListenerContainerOptions);
        streamMessageListenerContainer.receive(Consumer.from(groupName, consumer),
                StreamOffset.create(consumer, ReadOffset.lastConsumed()), this.streamMessageListener);
        this.streamMessageListenerContainer = streamMessageListenerContainer;
        this.streamMessageListenerContainer.start();
    }

    @Override
    public void destroy() throws Exception {
        this.streamMessageListenerContainer.stop();
    }
}

and this is the StreamMessageListener:

package com.dolphin.soa.post.common.mq;

import com.dolphin.soa.post.contract.request.ArticleRequest;
import com.dolphin.soa.post.model.entity.SubRelation;
import com.dolphin.soa.post.service.IArticleService;
import com.dolphin.soa.post.service.ISubRelationService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.DefaultTypedTuple;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;

/**
 * @author dolphin
 */
@Component
@Slf4j
public class StreamMessageListener implements StreamListener<String, MapRecord<String, String, String>> {

    @Value("${dolphin.redis.stream.group}")
    private String groupName;

    @Value("${dolphin.redis.user.sub.article.key}")
    private String subArticleKey;

    private final StringRedisTemplate stringRedisTemplate;

    private final RedisTemplate<String, Object> articleRedisTemplate;

    private final RedisTemplate<String, Long> redisLongTemplate;

    private final ISubRelationService subRelationService;

    private final IArticleService articleService;

    public StreamMessageListener(StringRedisTemplate stringRedisTemplate,
                                 @Qualifier("redisObjectTemplate") RedisTemplate<String, Object> articleRedisTemplate,
                                 ISubRelationService subRelationService,
                                 @Qualifier("redisLongTemplate") RedisTemplate<String, Long> redisLongTemplate,
                                 IArticleService articleService) {
        this.stringRedisTemplate = stringRedisTemplate;
        this.articleRedisTemplate = articleRedisTemplate;
        this.subRelationService = subRelationService;
        this.redisLongTemplate = redisLongTemplate;
        this.articleService = articleService;
    }

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        try {
            RecordId messageId = message.getId();
            Map<String, String> body = message.getValue();
            log.info("stream message。messageId={}, stream={}, body={}", messageId, message.getStream(), body);
            handleArticle(body);
            this.stringRedisTemplate.opsForStream().acknowledge(groupName, message);
        } catch (Exception e) {
            log.error("error", e);
        }
    }
}

I am searching from internet and find no useful suggestions. I have tried to upgrade the lettuce package version to newest 6.1.1.RELEASE version but still not fix it. Where is the problem and What should I do to make it work as expect? I also tried replace lettuce with jedis:

api ("org.springframework.boot:spring-boot-starter-data-redis") {
            exclude group: "io.lettuce", module: "lettuce-core"
}
api "redis.clients:jedis:3.6.0"

seems it did not work.

Answer 1

For those using Maven instead of Gradle, and following Dolphin's answer, you ought to exclude lettuce-core artifact in your pom.xml file:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
    <exclusions>
        <exclusion>
            <groupId>io.lettuce</groupId>
            <artifactId>lettuce-core</artifactId>
        </exclusion>
    </exclusions>
</dependency>

And add jedis Maven dependency:

<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
</dependency>

---

Note: If you're using Spring Boot 2.5.x, then the version of spring-boot-starter-data-redis will require jedis version 3.9.0. This is due to the breaking changes introduced when moving from Jedis 3 to Jedis 4 (see this answer).

Answer 2

Check your Redis connections from the application. Make sure established connections have keepalive on them.

netstat -ton | grep ESTABLISHED | grep 6379

tcp 0 0 IP:39430 :6379 ESTABLISHED off (0.00/0/0)

change it to:

tcp 0 0 IP:39430 :6379 ESTABLISHED keepalive

Answer 3

For those who want to continue using 'spring-data-redis':

Gradle config:

api("org.springframework.boot:spring-boot-starter-data-redis"){
    exclude(group = "io.lettuce", module = "lettuce-core")
}
implementation("redis.clients:jedis:4.2.3")
implementation("org.apache.commons:commons-pool2:2.0")

Answer 4

I tried to using jedis not using lettuce solve this problem.

step 1: exclude lettuce:

api ("org.springframework.boot:spring-boot-starter-data-redis") {
            exclude group: "io.lettuce", module: "lettuce-core"
            exclude group: "org.springframework.data", module: "spring-data-redis"
        }

because the spring-boot-starter-data-redis contains both jedis and lettuce, if you exclude lettuce, the default connection should use jedis.