Connection exception during topology update

[suxb201]

Bug Report

I encountered an issue with Lettuce when sending commands while it updates the cluster topology, leading to exceptions. In my test, I continuously called Lettuce's read and write commands while performing a master-slave switch in a Redis cluster. Occasionally, Lettuce throws a connection exception, which appears as if the connection for normal commands was forcibly terminated.

Current Behavior

In most cases, no exceptions are thrown. However, there is a low probability of encountering the following exceptions:

• io.lettuce.core.RedisCommandTimeoutException: Command timed out after 1 minute(s)
• io.lettuce.core.RedisException: Connection closed

io.lettuce.core.RedisException: Connection closed

io.lettuce.core.RedisException: Connection closed
        at io.lettuce.core.protocol.DefaultEndpoint.lambda$notifyDrainQueuedCommands$7(DefaultEndpoint.java:677)
        at io.lettuce.core.protocol.DefaultEndpoint$Lazy.getNullable(DefaultEndpoint.java:1182)
        at io.lettuce.core.protocol.DefaultEndpoint$Lazy.get(DefaultEndpoint.java:1167)
        at io.lettuce.core.protocol.DefaultEndpoint.lambda$notifyDrainQueuedCommands$8(DefaultEndpoint.java:678)
        at io.lettuce.core.protocol.DefaultEndpoint.cancelCommands(DefaultEndpoint.java:799)
        at io.lettuce.core.protocol.DefaultEndpoint.notifyDrainQueuedCommands(DefaultEndpoint.java:678)
        at io.lettuce.core.protocol.CommandHandler.channelInactive(CommandHandler.java:368)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274)
        at io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:81)
        at io.lettuce.core.protocol.RedisHandshakeHandler.channelInactive(RedisHandshakeHandler.java:79)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274)
        at io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:81)
        at io.lettuce.core.ChannelGroupListener.channelInactive(ChannelGroupListener.java:54)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274)
        at io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1352)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:301)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
        at io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:850)
        at io.netty.channel.AbstractChannel$AbstractUnsafe$7.run(AbstractChannel.java:811)
        at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
        at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasksFrom(SingleThreadEventExecutor.java:428)
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:377)
        at io.netty.incubator.channel.uring.IOUringEventLoop.run(IOUringEventLoop.java:222)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.base/java.lang.Thread.run(Thread.java:1575)

io.lettuce.core.RedisCommandTimeoutException: Command timed out after 1 minute(s)

io.lettuce.core.RedisCommandTimeoutException: Command timed out after 1 minute(s)
    at io.lettuce.core.internal.ExceptionFactory.createTimeoutException(ExceptionFactory.java:63)
    at io.lettuce.core.internal.Futures.awaitOrCancel(Futures.java:233)
    at io.lettuce.core.cluster.ClusterFutureSyncInvocationHandler.handleInvocation(ClusterFutureSyncInvocationHandler.java:136)
    at io.lettuce.core.internal.AbstractInvocationHandler.invoke(AbstractInvocationHandler.java:84)
    at jdk.proxy6/jdk.proxy6.$Proxy67.set(Unknown Source)
    at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
    at java.base/java.lang.reflect.Method.invoke(Method.java:580)
    at io.lettuce.core.support.ConnectionWrapping$DelegateCloseToConnectionInvocationHandler.handleInvocation(ConnectionWrapping.java:200)
    at io.lettuce.core.internal.AbstractInvocationHandler.invoke(AbstractInvocationHandler.java:84)
    at jdk.proxy6/jdk.proxy6.$Proxy67.set(Unknown Source)
    at tair.Main.main(Main.java:47)
    at org.codehaus.mojo.exec.ExecJavaMojo.lambda$execute$0(ExecJavaMojo.java:283)
    at java.base/java.lang.Thread.run(Thread.java:1575)

Packet Capture Analysis

io.lettuce.core.RedisCommandTimeoutException: Command timed out after 1 minute(s):

1. The main thread calls Lettuce to send a SET command.
2. While waiting for the SET response, Lettuce completes the topology update and sends a FIN to the old connection.
3. The database replies with MOVED, and Linux directly sends a RST.
4. Lettuce waits for the SET command response until a timeout occurs.

Additionally, it is speculated that another exception, io.lettuce.core.RedisException: Connection closed, may follow this sequence:

1. The main thread calls Lettuce to send a SET command, but the command is not sent.
2. Lettuce completes the topology update and sends a FIN to the old connection.
3. The actual SET packet is sent, resulting in a Connection closed error.

Steps to Reproduce

1. Continuously execute Lettuce read and write commands.
2. Perform a master-slave switch in the Redis cluster.
3. Observe the connection exceptions thrown by Lettuce.

Input Code

package tair;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

import io.lettuce.core.RedisURI;
import io.lettuce.core.SocketOptions;
import io.lettuce.core.SocketOptions.KeepAliveOptions;
import io.lettuce.core.SocketOptions.TcpUserTimeoutOptions;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.cluster.ClusterClientOptions;
import io.lettuce.core.cluster.ClusterTopologyRefreshOptions;
import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.protocol.ProtocolVersion;
import io.lettuce.core.support.ConnectionPoolSupport;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

public class Main {

    private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");

    public static void main(String[] args) throws InterruptedException {
        String host = "xxxxxxxx.redis.rds.aliyuncs.com";
        int port = 6379;

        RedisURI redisURI = RedisURI.Builder.redis(host).withPort(port).build();

        ClusterTopologyRefreshOptions refreshOptions = ClusterTopologyRefreshOptions.builder()
                .enablePeriodicRefresh(Duration.ofSeconds(3000000)).dynamicRefreshSources(true)
                .enableAllAdaptiveRefreshTriggers().adaptiveRefreshTriggersTimeout(Duration.ofSeconds(20)).build();

        RedisClusterClient redisClient = RedisClusterClient.create(redisURI);
        redisClient.setOptions(ClusterClientOptions.builder().protocolVersion(ProtocolVersion.RESP2)
                .validateClusterNodeMembership(false).topologyRefreshOptions(refreshOptions).build());

        // GenericObjectPool<StatefulRedisClusterConnection<String, String>> pool = ConnectionPoolSupport
        //         .createGenericObjectPool(() -> redisClient.connect(), new GenericObjectPoolConfig());
        StatefulRedisClusterConnection<String, String> connection = redisClient.connect();
        for (int i = 0; i < 1000000; i++) {
            try {
                // StatefulRedisClusterConnection<String, String> connection = pool.borrowObject();
                long begin = System.currentTimeMillis();
                for (int j = 0; j < 5000; j++) {
                    connection.sync().set("" + i + j, "" + i + j);
                    connection.sync().get("" + i + j);
                }
                long end = System.currentTimeMillis();
                String now = LocalDateTime.now().format(formatter);
                System.out.println("now: " + now + ",i: " + i + ", cost: " + (end - begin) + "ms");
                // pool.returnObject(connection);
            } catch (Exception e) {
                System.out.println("\nException throwed!\n");
                e.printStackTrace();
            }
        }

    }

}

Expected behavior/code

No exceptions.

Environment

• Lettuce version(s): the main branch
• Redis version: Redis 7.0

[tishun]

Hey @suxb201 ,

Could you comment on what your expectation is for the driver to handle these cases?

[suxb201]

@tishun I want the client to handle cluster topology changes without throwing any exceptions, as exceptions are difficult to handle and retrying non-idempotent commands on exceptions can lead to data inconsistencies. We can achieve this by fully utilizing the MOVED response during topology changes:

1. If the client accesses the correct node, it returns the result normally.
2. If the client accesses the wrong node, it receives a MOVED response to redirect to the correct node and updates the topology.
3. Nodes scheduled for shutdown remain active for a while to allow clients to complete redirections.

In practice, most clients(Jedis, redis-py, etc.) remain unaffected during cluster topology changes, experiencing only slight increases in latency. Lettuce is an exception; even with the correct configuration, it sometimes throws exceptions, which I believe is a bug related to how Lettuce handles connection closures.

[tishun]

Let me try to address all the things you said separately.

We can achieve this by fully utilizing the MOVED response during topology changes:

I am really confused by your example, because Lettuce always reacts to a MOVED response. In your initial description you mention that there is a command waiting for a response - during the topology update - that eventually times out. If it times out then it never receives a response from the server (MOVED is a valid response that the driver can and will handle) in the first place. So the problem does not seem to be related to not handling a MOVED response, but instead not knowing what to do when the topology changed during sending a command.

This is why I asked what do you think we should do - if the command was sent during a transitional state of the system and we can't know if ...

• a) it was executed successfully or not
• b) it was executed against the right node or not
• c) it should be retried or ignored
  ... so this is left to the user application to handle (by utilizing a try-catch block and implementing some fallback logic)

Important

Please have in mind that the topology updates are being sent on a separate connection and are sent asynchronously from any other connection / connections that the user application might have established. Lettuce in also a non-blocking client. When you use this pattern you need to accept the fact that there would be events that would be happening in the same time, for example cluster topology refresh and command execution.

In practice, most clients(Jedis, redis-py, etc.) remain unaffected during cluster topology changes, experiencing only slight increases in latency. Lettuce is an exception; even with the correct configuration, it sometimes throws exceptions, which I believe is a bug related to how Lettuce handles connection closures.

Jedis is a synchronous client, while Lettuce is asynchronous. Lettuce also allows several threads to multiplex the same connection. So it is a bit of an odd comparison to make in the first place, because the problem you are describing does not really exist in the first place in Jedis. Depending on how you use redis-py this might be the case there too.

In conclusion

All that said if there is a reasonable way we can improve the resilience of the driver during topology refresh I am more than happy to discuss it. We need to identify the specific problem and discuss a solution to it that would fit the common use cases of the driver.

If you have evidence that the driver is not reacting to a MOVED response, please let me know how I can reproduce that.

[tishun]

Let me see if we can spend some more time thinking about how we can improve driver resilience here.

PRs / suggestions are welcome.

[tishun]

Currently unable to reproduce, using the code provided.

Perform a master-slave switch in the Redis cluster.

What steps do you take in order to achieve that? Calling CLUSTER FAILOVER on the replica?

[suxb201]

What steps do you take in order to achieve that? Calling CLUSTER FAILOVER on the replica?

We made many changes to the Redis Cluster. Some components will directly send the topology table to the database. The database will then create connections on its own based on the information in the topology table.

So the problem does not seem to be related to not handling a MOVED response, but instead not knowing what to do when the topology changed during sending a command.

I completely agree with your view. It was my description that caused the misunderstanding.

In my view, when modifying the Lettuce topology, we should properly handle active connections. For connections that need to be closed but have pending requests, it's crucial to wait for the response before closing the connection. We shouldn't close the connection before receiving the response, as this would prevent us from ever getting that response:

I have hardly written any Java code before, and reading asynchronous Java code is quite challenging. It's difficult for me to submit a PR (Pull Request).

[tishun]

Understood.

I have hardly written any Java code before, and reading asynchronous Java code is quite challenging. It's difficult for me to submit a PR (Pull Request).

This is completely fine. Sometimes it is better to communicate ideas with code, but in this case we can try and figure this out another way.

We made many changes to the Redis Cluster. Some components will directly send the topology table to the database. The database will then create connections on its own based on the information in the topology table.

This is where my problem currently stands. To decide how to handle these I need to reproduce the problem and analyse possible solutions, otherwise I would be making changes that might or might not help.

In my view, when modifying the Lettuce topology, we should properly handle active connections. For connections that need to be closed but have pending requests, it's crucial to wait for the response before closing the connection. We shouldn't close the connection before receiving the response, as this would prevent us from ever getting that response:

I agree. However I am not sure why the connection is closed. I was unable to reproduce it with the sample code you sent, because something in the way you do the failover causes it.

I will try out a few other things, but until we reproduce it this would be hard to fix properly.