Lettuce/Lobby

Chat room for the Lettuce Redis driver project

Le Xu

@flint-stone

Hello! I have a question about using lettuce async operation to achieve iterative check-and-set operation -- basically I have a single-threaded client that basically does this 'commands = connection.async();
while(true){
commands.get(key).thenAccept(value -> commands.set(key, value+1));
}' But as you could see using an async client let multiple read operations (which returns the same value) to carry out before add operation. If I'd like to achieve the atomic check-and-set operation described here, is it a good practice to use WATCH and MULTI even when I only have one client? I wasn't able to find an example online except the Using WATCH example here. But it wasn't clear to me how MULTI could be applied to the loop with async client. ( I do realize that lettuce supports INCR but I'm only wondering what's the best practice to achieve atomic check-and-set operation with async client)

commands = connection.async();
while(true){
commands.get(key).thenAccept(value -> commands.set(key, value+1));
}

Le Xu

@flint-stone

I also realize that using synchronized client can resolve this but I'm just curious whether there is any way to achieve this with asynchronous client :)

Le Xu

@flint-stone

What is the difference between channelWriteAndFlush() in DefaultEnpoint and CommandHandler.write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) -- I can see both of them use netty ChannelOutputInvoker but I can see all my commands walk through both functions before sent and executed

pottlapelli

@pottlapelli

Hi All, DefaultClusterTopologyRefresh is reporting "No route to host", not sure how it gets ip address 10.244.8.126. I log the partition info on a successful connection and it is not one of the nodes in Redis cluster. Please see the logs below, what could be going wrong? I appreciate any help on debugging/fix the issue.

{
"level": "INFO",
"logger": "oracle.oic.v3.mtms.infra.cache.RedisConnection",
"thread": "io-executor-thread-6",
"ts": 1616648819527,
"msg": "connection successful Cluster info:\nRedisClusterNodeSnapshot [uri=redis://10.244.9.226, nodeId='356ed4cc2feb77e5bef1ed03106c39e5621dcc75', connected=true, slaveOf='af6deffcfc98c616df752f65663198fd06034fcf', pingSentTimestamp=0, pongReceivedTimestamp=1616648816515, configEpoch=12, flags=[SLAVE, REPLICA], aliases=[]]\nRedisClusterNodeSnapshot [uri=redis://10.244.8.66, nodeId='2f7c5f80a6cbc1b31d03d6cacf150b012624fcda', connected=true, slaveOf='81c972f74906b648b97a34f7daac0d3458ddf235', pingSentTimestamp=0, pongReceivedTimestamp=1616648818519, configEpoch=11, flags=[SLAVE, REPLICA], aliases=[]]\nRedisClusterNodeSnapshot [uri=redis://10.244.7.248, nodeId='ef0f40f54c8375d05cc264f6d91e2b70d805624a', connected=true, slaveOf='e9513a9553fa196307cb8834f9d235a175641629', pingSentTimestamp=0, pongReceivedTimestamp=1616648816000, configEpoch=10, flags=[MYSELF, SLAVE, REPLICA], aliases=[redis://10.244.7.248]]\nRedisClusterNodeSnapshot [uri=redis://10.244.9.67, nodeId='81c972f74906b648b97a34f7daac0d3458ddf235', connected=true, slaveOf='null', pingSentTimestamp=0, pongReceivedTimestamp=1616648817518, configEpoch=11, flags=[MASTER], aliases=[], slot count=5462]\nRedisClusterNodeSnapshot [uri=redis://10.244.6.137, nodeId='e9513a9553fa196307cb8834f9d235a175641629', connected=true, slaveOf='null', pingSentTimestamp=0, pongReceivedTimestamp=1616648817000, configEpoch=10, flags=[MASTER], aliases=[], slot count=5461]\nRedisClusterNodeSnapshot [uri=redis://10.244.5.181, nodeId='af6deffcfc98c616df752f65663198fd06034fcf', connected=true, slaveOf='null', pingSentTimestamp=0, pongReceivedTimestamp=1616648816000, configEpoch=12, flags=[MASTER], aliases=[], slot count=5461] dnsResolver UNRESOLVED",
"oic.partition": "1",
"opc-request-id": "A2FF54E7DE9DCD2BA5329A0AAE35E921/4E78AAD01904A5CA6D12FC1005638711/7ADBF930AC9DEFEC883CF509A2639397",
"oic.service-instance": "ocid1.integrationinstance.oc1.phx.mpottlap",
"oic.trace": "8e4cce6096e70df155c2c0d0574d443c",
"oic.user": "ocid1.integrationinstance.oc1.phx.mpottlap-admin",
"timestamp": "2021-0324 @ 22:06:59"
}
{
"level": "WARN",
"logger": "io.lettuce.core.cluster.topology.DefaultClusterTopologyRefresh",
"thread": "lettuce-nioEventLoop-6-2",
"ts": 1616648880381,
"msg": "Unable to connect to [10.244.8.126:6379]: No route to host: /10.244.8.126:6379",
"timestamp": "2021-0324 @ 22:08:00"
}

Mark Paluch

@mp911de

Lettuce uses CLUSTER NODES to obtain details of the Redis Cluster topology

2 replies

bschup

@bschup

HI Folks!

I have been playing around with lettuce/redis for a little bit and may run into an issue.
For the very simple pubsub subscriber below, I seem to be missing a few bytes from the delivered message. I expect 1024 btyes and if I check the len() of the message and the results of .getBytes() I'm consistenly short by 10-15 bytes. I verified that the publisher has delivered the entire payload to redis via "publish" with a similaryly simple python client (everything checks out, proper length and values).

Not sure where to start looking, I would expect that the Listener is pulling all the bytes from the network, is there a mechanism to flush any buffers involved in the process? At the end of the day I would like to convert the 1024 bytes to floats (which I can successfully do from the python client).

using lettuce 6.0
thanks

RedisURI builder = RedisURI.Builder
.redis(MY_IP, PORT)
.withAuthentication("", PASSWORD)
.withDatabase(0).build();

RedisClient client = RedisClient.create(builder);

// subscribe to a waveform channel, chart floats coming from C++ client/producer
StatefulRedisPubSubConnection<String, String> psconnection2 = client2.connectPubSub();
RedisPubSubListener<String, String> listener2 = new RedisPubSubAdapter<String, String>() {
@Override
public void message(String channel, String message) {

        // parse the message
        byte[] bs = message.getBytes();
    // do some stuff with the message
    }

};

psconnection2.addListener(listener2);
RedisPubSubAsyncCommands<String, String> psasync2 = psconnection2.async();
RedisFuture<Void> pfuture2 = psasync2.subscribe(":waveform");

try {
pfuture2.get();
} catch () {
// handle errors
}

p4w4n

@p4w4n

@mp911de Hi Mark, We are using redis in our production environment and lettuce as our connection API. After running the application for around 2 weeks we saw a surge in heap memory usage. I took the heap dump for the application and used MemoryAnalyzer to analyze it. the leak detector pointed out org.LatencyUtils.SimplePauseDetector Thread as a potential leak as it is holding the maximum heap. Below is a snapshot if it can be of any help:

One instance of "org.LatencyUtils.SimplePauseDetector" loaded by "org.springframework.boot.loader.LaunchedURLClassLoader @ 0x80017380" occupies 329,570,400 (60.31%) bytes. The memory is accumulated in one instance of "java.lang.Object[]", loaded by "<system class loader>", which occupies 329,568,744 (60.31%) bytes.

Keywords
org.LatencyUtils.SimplePauseDetector
org.springframework.boot.loader.LaunchedURLClassLoader @ 0x80017380
java.lang.Object[]

Mark Paluch

@mp911de

That's indeed a lot. Are you using custom commands? That is an implementation of ProtocolKeyword in your code? It would make sense to dive into normalPriorityListeners/highPriorityListeners of PauseDetector and inspect CommandLatencyId why there are so many instances of LatencyStats retained.

p4w4n

@p4w4n

@mp911de No, we don't have any implementation of ProtocolKeyword in our code. We are using redis 5.3 and io.lettuce-core 5.3.2-RELEASE. Redis runs as a cluster with 3 master nodes and 2 replica nodes for each master. Although, instead of using spring-data-redis, we created our own interface with the redis using the lettuce APIs. We have also observed that when a node goes down and comes up again, somehow the older connections are not released and new connections are created to the node again. This results in a lot of unused connections. We have a plan to upgrade to redis 6 and new lettuce API version, but this will take some months.
Is it possible to disable the LatencyStats altogether.

Mark Paluch

@mp911de

Yes, it is. Just configure via CommandLatencyCollector.disabled() through ClientResources.

6 replies

p4w4n

@p4w4n

Thanks a lot Mark, I will try that. Thanks again for your help. :)

pottlapelli

@pottlapelli

Lettuce uses CLUSTER NODES to obtain details of the Redis Cluster topology

Mark, thanks for the response. IP address (10.244.8.126) printed by DefaultClusterTopologyRefresh is not one of the IP addresses in the first log statement. And WARN message does not stop, what could be wrong? how to debug?

Dmitry Kotov

@kotovdv

Hi everyone.

Could someone help me to understand if Lettuce can track topology changes among sentinels?
I know that topology changes among redis nodes themselves are tracked, but I was unable to find the same functionality for sentinels.

For example:
I have N sentinels deployed on k8s.
If during start up I provide lettuce at least 1 ip address of working sentinel - I'd really like to see other sentinels to be discovered automatically.

2 replies

xdu

@duxing

hi everyone,
I'm using AWS ElasticCache Redis (6.x, cluster mode on) with latest version of lettuce-core (6.1.0-RELEASE)
I was wondering if anyone has tried specifying ioThreadPoolSize and computationThreadPoolSize in DefaultClientResources.Builder

the default value is number of available processor, in my case is 4 (the instance where I'm running my application). and after I bumped this number to 8 and recreate a cluster client, I do see 8 threads from attached jprofiler, but from the application logs (trace level), only 3 threads are involved. ("lettuce-epollEventLoop-*"). Note that when the io thread size is not specified (default to 4), the same number of threads (3) are involved in logs

seems to me that threads are being created but not being used by lettuce somehow. has anyone looked into this before?

xdu

@duxing

update on my question:

I've turned on trace log for all my redis related code as well as io.lettuce.core package.
ioThreadPoolSize is configured to 8 and computationThreadPoolSize is configured to 2
collected logs and aggregated by thread name, and I found that all 8 io threads (lettuce-epollEventLoop-12-[1-8]) are involved in cluster topology refresh (a burst of logs from AbstractRedisClient,ClusterTopologyRefreshScheduler,CommandEncoder,CommandHandler,ConnectionWatchdog,DefaultEndpoint,PooledClusterConnectionProvider,RedisChannelHandler,RedisStateMachine for every 5 seconds, as I configured)
while my application trace log (added in thenApply, thenAccept or whenComplete of returned RedisFuture) contains only 3 unique threads (lettuce-epollEventLoop-12-[1-3] )

as for the log volume by thread name, all the topology fresh logs are evenly distributed across 8 threads, and all application trace logs are evenly distributed on 3 threads.

I've kept the application running for over an hour with tens of thousands of different types of redis commands (sismember, expire, get, mget etc) per minute to aggregate the logs above.

@mp911de do you know what could possibly lead to this? I feel I'm one configuration away from getting where I want to be.

Thanks in advance!

xdu

@duxing

^ seems like the above issue is caused by number of connections. in my case I;m using a cluster with 3 shards and in total 3 connection to each shards. I saw somewhere in lettuce-core documentation about 1 thread per connection and verified that by adding connection pooling: when maximum idle connection count is more than 8, all 8 threads surfaced in the log

Srthak100

@Srthak100

Hi @mp911de , we are facing some intermittent behaviour in reconnection to the redis cluster with lettuce library.

We have a microservice which uses lettuce to connect to redis cluster.

We are first removing the service from the network and then getting it back to the network. At first the connection to every middleware breaks and gets reconnected, but in case of redis cluster we are not getting reconnected even after a span of 10 min. We have autoreconnect=true, periodic and adaptive refresh enabled.

Srthak100

@Srthak100

Does topology refresh has any relation to tcp connections? Also is there any particular time period between reconnect attempts? Also is there any other property which enables lettuce to reconnect to redis cluster other than autoreconnect?

Please let me know if you want anything more to advice on this issue. @mp911de

1 reply

Srthak100

@Srthak100

Lettuce version:5.2.2

Redis version:5.7

Redis arch: 5M-5S

Wei Gao

@Noodle05

Hi Team, I'm little confused by async API and transaction. From the example: it just issue cmds.multi(), cmds.set..., cmds.set... and then cmds.exec. And wait cmds.exec's future complete. Shouldn't we chain them like after cmds.multi complete, issue cmds.set, after first cmds.set complete issue second cmds.set, and all complete issue cmds.exec?

Mark Paluch

@mp911de

You can use the API either way as the asynchronous API respects the order of commands when calling API methods sequentially from the same thread.

garramja

@garramja

Hi. Somebody has configured lettuce client with springboot to connect AWS redis elasticcache master/slave?
I have an aws elasticcache master/slaves and I'am able to write and read but a error is thrown every 5000ms
The service works fine but I want to remove the error or how to know to fix it.

This is the console error ()

"level":"ERROR","className":"org.springframework.data.redis.listener.RedisMessageListenerContainer","message":"Connection failure occurred. Restarting subscription task after 5000 ms "

But this is the original error from RedisMessageListenerContainer.handleSubscriptionException

java.lang.UnsupportedOperationException: Pub/Sub connections not supported with Master/Replica configurations

This is the bean configuration

    @Bean
    public LettuceConnectionFactory redisConnectionFactory() {

        LettuceClientConfiguration.LettuceClientConfigurationBuilder lettuceConfigBuilder = LettuceClientConfiguration.builder()
                .readFrom(ReadFrom.REPLICA_PREFERRED);
        lettuceConfigBuilder.useSsl();

        RedisStaticMasterReplicaConfiguration redisConfiguration = new RedisStaticMasterReplicaConfiguration(master, port);
        redisConfiguration.addNode(replica, port);
        redisConfiguration.setPassword(password);

        return new LettuceConnectionFactory(redisConfiguration, lettuceConfigBuilder.build());
    }

Mark Paluch

@mp911de

How is Pub/Sub supposed to work across replicas? Redis Cluster has a mechanism to propagate messages into the entire cluster. Redis Standalone (Master/Replica) doesn't do this.

garramja

@garramja

Yep. With cluster mode the error dissapears but maybe pub/sub with master/replica should have flag to disable it

Victor Babenko

@virusman

Hi! Does pipelining support evalsha, or is every evalsha command blocking?

Mark Paluch

@mp911de

It does support pipelining. Lettuce pipelines all commands.

Peter Jurkovic

@peterjurkovic

Hello, I trying to figure out how can I use WATCH (optimistic locking) with Reactive client, it is possible? I don't see any "multi()" option.

Mark Paluch

@mp911de

Transactions using the reactive API can get tricky because of how Redis responds. These two aren't actually compatible. Using the async API and converting the result of exec into a Mono<TransactionResult> is IMO the smoother and cleaner option.

Peter Jurkovic

@peterjurkovic

Could you please explain why Reactive API is not compatible with Redis transactions?

What I had in mind it following:

RedisReactiveCommands<String, String> reactive = conn.reactive();

Mono<?> result = reactive.watch("key")
                         .flatMap(__ -> reactive.get("key"))
                         .map(value -> value + " hi")
                         .flatMap( newValue -> reactive.multi()
                                                       .flatMap(__ -> reactive.set("key", newValue))
                                                       .flatMap(__ -> reactive.exec()));

but looks like reactive.set() got stuck for some reason

Mark Paluch

@mp911de

Exactly that. Redis Transactions incorporate two steps for each command. Sending a command during an active transaction lets Redis respond with QUEUED. If you translate that to a Java API, then the command has been sent but not yet completed. It would feel weird, to respond with QUEUED on each command. Therefore, commands are pending upon subscription. They only complete after invoking EXEC

3 replies

Shashidhar

@shashdr

Hi , Seeing this issue with Lettuceconnectionfactory connecting to Redis server,

org.springframework.data.redis.RedisConnectionFailureException: Unable to connect to Redis; nested exception is io.lettuce.core.RedisConnectionException: Unable to connect to <redis server>:6379

caused by

Caused by: io.lettuce.core.RedisCommandExecutionException: unsupported command 'HELLO'
    at io.lettuce.core.internal.ExceptionFactory.createExecutionException(ExceptionFactory.java:137)
    at io.lettuce.core.internal.ExceptionFactory.createExecutionException(ExceptionFactory.java:110)

using io.lettuce:6.1.0-release, spring.data.redis:2.4.7 , Any idea what unsupported command meant by?

Mark Paluch

@mp911de

Lettuce 6 performs a protocol handshake to determine the Redis protocol version. I'm wondering why RedisCommandExecutionException: unsupported command 'HELLO' happens at all. IIRC, this should only happen if you configure Lettuce to use RESP3.

Shashidhar

@shashdr

Nope, I haven't configure to use RESP3 ,

    public LettuceConnectionFactory lettuceConnectionFactory(Environment environment, @Value("${redisServerIP:localhost}") String redisServerIP,
                                                           @Value("${redisServerPort:6379}") Integer redisServerPort, ClientOptions clientOptions) {
        RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration();
        System.out.println("redisServerIP" +redisServerIP);
        redisStandaloneConfiguration.setHostName(redisServerIP);
        redisStandaloneConfiguration.setPort(redisServerPort);
        LettuceClientConfiguration.LettuceClientConfigurationBuilder lettuceClientConfigurationBuilder = LettuceClientConfiguration.builder()
                .clientOptions(clientOptions);
        return new LettuceConnectionFactory(redisStandaloneConfiguration, lettuceClientConfigurationBuilder.build());
    }

So, here's the thing the TLS handshake happens via envoy that resides as side car container, and envoy has redis-proxy enabled which does outbound calls, in this case calls to redis server happen through envoy.

Shashidhar

@shashdr

Do you suggest downgrading to RESP2 ?

Mark Paluch

@mp911de

Yes

Shashidhar

@shashdr

HELLO is sent when Lettuce tries to auto-discover the protocol. You can switch to RESP2 in ClientOptions.

I have seen your comments here and will try it out and update, Thank you Mark

Yeah Now I see this error
Error in execution; nested exception is io.lettuce.core.RedisCommandExecutionException: NOAUTH Authentication required

Shashidhar

@shashdr

any other suggestions for the above error @mp911de I see this issue with AWS elasticache StackExchange/StackExchange.Redis#1273

rreddy13Official

@rreddy13Official

Hi , I am using lettuce(6.0.1.RELEASE) to connect to Redis 6.0.9 and enabled client side caching. Everything works fine as long as the connection is active. When connection is interrupted due to network, it tries to reconnect. After successful reconnection, the changes to the redis are not invalidated to the client(local) cache. Looks like the Client side caching is not applied to the connection upon reconnect. OR am i missing anything ? How could i solve this so that the client side caching works fine even after reconnect ? Is there any option to flush the local cache when connection is interrupted ?Thanks in advance.

Mark Paluch

@mp911de

@rreddy13Official The problem with disconnects is that after a reconnect, the touched cache keys are no longer known to Redis because a reconnect creates a new connection. I don't have a good idea yet. A workaround could be GET'ing all the touched keys by the client but that would affect LRU caching.

rreddy13Official

@rreddy13Official

@mp911de - The major problem we have now is that client cache has stale data which need to be flushed. Any work around for that ?

Mark Paluch

@mp911de

Lettuce provides a frontend with cache accessors to wrap an existing cache and augment it with Redis functionality. I suggest flushing the underlying client-side cache to work around that problem. You can either register connection state listeners via RedisClient or listen to the EventBus for disconnect events to get notified of disconnects.

rreddy13Official

@rreddy13Official

@mp911de - Sounds like the option that I am looking for. I will try the option . One more question - Does this new connection has the client side cache enabled like the initial connection ? Or I need to apply it through the listener/event ?

Where communities thrive

People

Repo info

Activity