how to refresh lettuce connection manually?

[carresia]

My project need to refresh redis nodes manually when redis config changed（we already enable periodic refresh also）.
How to manually refresh?

The follow is my answer, but there maybe some wrong:

lettuceConnectionFactory.destroy();
            // because destory is async, so, wait a time...
            Thread.sleep(3000);

            // reset redisNodes
            List<RedisNode> curRedisNodes = curHostAndPort.stream().map(hp -> {
                String[] hostAndPort = hp.split(":");
                return new RedisNode(hostAndPort[0], Integer.valueOf(hostAndPort[1]));
            }).collect(Collectors.toList());
            lettuceConnectionFactory.getClusterConfiguration().setClusterNodes(curRedisNodes);
            String password = (String) properties.get(redisPassword);
            lettuceConnectionFactory.getClusterConfiguration().setPassword(RedisPassword.of(password));

            **lettuceConnectionFactory.afterPropertiesSet();
            lettuceConnectionFactory.validateConnection();**
            log.info("lettuceConnectionFactory already validateConnection...");`


The following is error log:

`
// start....
    at org.apache.dubbo.remoting.transport.dispatcher.ChannelEventRunnable.run(ChannelEventRunnable.java:57)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: io.lettuce.core.RedisConnectionException: Unable to connect to [RedisURI [host='...', port=41837], RedisURI [host='...', port=41838]...]
    at io.lettuce.core.RedisConnectionException.create(RedisConnectionException.java:78)
    at io.lettuce.core.cluster.RedisClusterClient.lambda$transformAsyncConnectionException$32(RedisClusterClient.java:1136)
    at io.lettuce.core.DefaultConnectionFuture.lambda$thenCompose$1(DefaultConnectionFuture.java:252)
    at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760)
    at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736)
    at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
    at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1977)
    at reactor.core.publisher.MonoToCompletableFuture.onError(MonoToCompletableFuture.java:68)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.lambda$onError$1(TracingSubscriber.java:63)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.withActiveSpan(TracingSubscriber.java:79)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.onError(TracingSubscriber.java:63)
    at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onError(FluxMapFuseable.java:134)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.lambda$onError$1(TracingSubscriber.java:63)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.withActiveSpan(TracingSubscriber.java:79)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.onError(TracingSubscriber.java:63)
    at reactor.core.publisher.FluxHide$SuppressFuseableSubscriber.onError(FluxHide.java:132)
    at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onError(FluxPeekFuseable.java:227)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.lambda$onError$1(TracingSubscriber.java:63)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.withActiveSpan(TracingSubscriber.java:79)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.onError(TracingSubscriber.java:63)
    at reactor.core.publisher.FluxHide$SuppressFuseableSubscriber.onError(FluxHide.java:132)
    at reactor.core.publisher.MonoFlatMap$FlatMapMain.onError(MonoFlatMap.java:165)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.lambda$onError$1(TracingSubscriber.java:63)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.withActiveSpan(TracingSubscriber.java:79)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.onError(TracingSubscriber.java:63)
    at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onError(FluxOnErrorResume.java:100)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.lambda$onError$1(TracingSubscriber.java:63)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.withActiveSpan(TracingSubscriber.java:79)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.onError(TracingSubscriber.java:63)
    at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onError(MonoPeekTerminal.java:251)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.lambda$onError$1(TracingSubscriber.java:63)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.withActiveSpan(TracingSubscriber.java:79)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.onError(TracingSubscriber.java:63)
    at reactor.core.publisher.FluxHide$SuppressFuseableSubscriber.onError(FluxHide.java:132)
    at reactor.core.publisher.MonoCompletionStage.lambda$subscribe$0(MonoCompletionStage.java:80)
    at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760)
    at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736)
    at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
    at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1977)
    at io.lettuce.core.AbstractRedisClient.lambda$initializeChannelAsync0$4(AbstractRedisClient.java:329)
    at io.opentelemetry.javaagent.instrumentation.netty.common.FutureListenerWrappers$WrappedFutureListener.operationComplete(FutureListenerWrappers.java:67)
    at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:577)
    at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:551)
    at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:490)
    at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:615)
    at io.netty.util.concurrent.DefaultPromise.setFailure0(DefaultPromise.java:608)
    at io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:117)
    at io.netty.bootstrap.Bootstrap.doResolveAndConnect0(Bootstrap.java:231)
    at io.netty.bootstrap.Bootstrap.access$000(Bootstrap.java:48)
    at io.netty.bootstrap.Bootstrap$1.operationComplete(Bootstrap.java:182)
    at io.netty.bootstrap.Bootstrap$1.operationComplete(Bootstrap.java:168)
    at io.opentelemetry.javaagent.instrumentation.netty.common.FutureListenerWrappers$WrappedFutureListener.operationComplete(FutureListenerWrappers.java:67)
    at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:577)
    at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:551)
    at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:490)
    at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:615)
    at io.netty.util.concurrent.DefaultPromise.setSuccess0(DefaultPromise.java:604)
    at io.netty.util.concurrent.DefaultPromise.trySuccess(DefaultPromise.java:104)
    at io.netty.channel.DefaultChannelPromise.trySuccess(DefaultChannelPromise.java:84)
    at io.netty.channel.AbstractChannel$AbstractUnsafe.safeSetSuccess(AbstractChannel.java:985)
    at io.netty.channel.AbstractChannel$AbstractUnsafe.register0(AbstractChannel.java:505)
    at io.netty.channel.AbstractChannel$AbstractUnsafe.access$200(AbstractChannel.java:416)
    at io.netty.channel.AbstractChannel$AbstractUnsafe$1.run(AbstractChannel.java:475)
    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:510)
    at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:388)
    at io.netty.util.concurrent.SingleThreadEventExecutor$6.run(SingleThreadEventExecutor.java:1044)
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    ... 1 common frames omitted
Caused by: java.lang.IllegalStateException: executor not accepting a task
    at io.netty.resolver.AddressResolverGroup.getResolver(AddressResolverGroup.java:60)
    at io.netty.bootstrap.Bootstrap.doResolveAndConnect0(Bootstrap.java:194)
    ... 22 common frames omitted


....

2023-04-24 16:19:05.396 [lettuce-eventExecutorLoop-3-3] [ERROR] i.n.u.concurrent.DefaultPromise.rejectedExecution - Failed to submit a listener notification task. Event loop shut down?
java.util.concurrent.RejectedExecutionException: event executor terminated
    at io.netty.util.concurrent.SingleThreadEventExecutor.reject(SingleThreadEventExecutor.java:981)
    at io.netty.util.concurrent.SingleThreadEventExecutor.offerTask(SingleThreadEventExecutor.java:388)
    at io.netty.util.concurrent.SingleThreadEventExecutor.addTask(SingleThreadEventExecutor.java:381)
    at io.netty.util.concurrent.SingleThreadEventExecutor.execute(SingleThreadEventExecutor.java:880)
    at io.netty.util.concurrent.DefaultPromise.safeExecute(DefaultPromise.java:841)
    at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:498)
    at io.netty.util.concurrent.DefaultPromise.addListener(DefaultPromise.java:183)
    at io.netty.channel.DefaultChannelPromise.addListener(DefaultChannelPromise.java:95)
    at io.netty.channel.DefaultChannelPromise.addListener(DefaultChannelPromise.java:30)
    at io.lettuce.core.AbstractRedisClient.initializeChannelAsync0(AbstractRedisClient.java:323)
    at io.lettuce.core.AbstractRedisClient.lambda$initializeChannelAsync$0(AbstractRedisClient.java:294)
    at reactor.core.publisher.LambdaMonoSubscriber.onNext(LambdaMonoSubscriber.java:168)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.lambda$onNext$0(TracingSubscriber.java:58)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.withActiveSpan(TracingSubscriber.java:79)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.onNext(TracingSubscriber.java:58)
    at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:203)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.lambda$onNext$0(TracingSubscriber.java:58)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.withActiveSpan(TracingSubscriber.java:79)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.onNext(TracingSubscriber.java:58)
    at reactor.core.publisher.FluxHide$SuppressFuseableSubscriber.onNext(FluxHide.java:127)
    at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onNext(MonoPeekTerminal.java:173)
    at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2148)
    at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.request(MonoPeekTerminal.java:132)
    at reactor.core.publisher.FluxHide$SuppressFuseableSubscriber.request(FluxHide.java:142)
    at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.request(FluxPeekFuseable.java:137)
    at reactor.core.publisher.LambdaMonoSubscriber.onSubscribe(LambdaMonoSubscriber.java:118)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.onSubscribe(TracingSubscriber.java:53)
    at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onSubscribe(FluxPeekFuseable.java:171)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.onSubscribe(TracingSubscriber.java:53)
    at reactor.core.publisher.FluxHide$SuppressFuseableSubscriber.onSubscribe(FluxHide.java:113)
    at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onSubscribe(MonoPeekTerminal.java:145)
    at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54)
    at reactor.core.publisher.Mono.subscribe(Mono.java:4087)
    at reactor.core.publisher.Mono.subscribeWith(Mono.java:4193)
    at reactor.core.publisher.Mono.subscribe(Mono.java:4059)
    at reactor.core.publisher.Mono.subscribe(Mono.java:3995)
    at reactor.core.publisher.Mono.subscribe(Mono.java:3967)
    at io.lettuce.core.AbstractRedisClient.initializeChannelAsync(AbstractRedisClient.java:289)
    at io.lettuce.core.cluster.RedisClusterClient.connectStatefulAsync(RedisClusterClient.java:705)
    at io.lettuce.core.cluster.RedisClusterClient.connectToNodeAsync(RedisClusterClient.java:499)
    at io.lettuce.core.cluster.RedisClusterClient$NodeConnectionFactoryImpl.connectToNodeAsync(RedisClusterClient.java:1171)
    at io.lettuce.core.cluster.topology.ClusterTopologyRefresh.getConnections(ClusterTopologyRefresh.java:263)
    at io.lettuce.core.cluster.topology.ClusterTopologyRefresh.loadViews(ClusterTopologyRefresh.java:74)
    at io.lettuce.core.cluster.RedisClusterClient.doLoadPartitions(RedisClusterClient.java:865)
    at io.lettuce.core.cluster.RedisClusterClient.loadPartitions(RedisClusterClient.java:845)
    at io.lettuce.core.cluster.RedisClusterClient.reloadPartitions(RedisClusterClient.java:784)
    at io.lettuce.core.cluster.ClusterTopologyRefreshScheduler$ClusterTopologyRefreshTask.doRun(ClusterTopologyRefreshScheduler.java:256)
    at io.lettuce.core.cluster.ClusterTopologyRefreshScheduler$ClusterTopologyRefreshTask.run(ClusterTopologyRefreshScheduler.java:236)
    at io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38)
    at io.netty.util.concurrent.PromiseTask.run(PromiseTask.java:73)
    at io.netty.util.concurrent.DefaultEventExecutor.run(DefaultEventExecutor.java:66)
    at io.netty.util.concurrent.SingleThreadEventExecutor$6.run(SingleThreadEventExecutor.java:1044)
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.lang.Thread.run(Thread.java:748)
2023-04-24 16:19:05.396 [lettuce-eventExecutorLoop-3-3] [WARN ] io.netty.channel.AbstractChannel - Force-closing a channel whose registration task was not accepted by an event loop: [id: 0x65f34f45]
java.util.concurrent.RejectedExecutionException: event executor terminated
    at io.netty.util.concurrent.SingleThreadEventExecutor.reject(SingleThreadEventExecutor.java:981)
    at io.netty.util.concurrent.SingleThreadEventExecutor.offerTask(SingleThreadEventExecutor.java:388)
    at io.netty.util.concurrent.SingleThreadEventExecutor.addTask(SingleThreadEventExecutor.java:381)
    at io.netty.util.concurrent.SingleThreadEventExecutor.execute(SingleThreadEventExecutor.java:880)
    at io.netty.channel.AbstractChannel$AbstractUnsafe.register(AbstractChannel.java:472)
    at io.netty.channel.SingleThreadEventLoop.register(SingleThreadEventLoop.java:87)
    at io.netty.channel.SingleThreadEventLoop.register(SingleThreadEventLoop.java:81)
    at io.netty.channel.MultithreadEventLoopGroup.register(MultithreadEventLoopGroup.java:86)
    at io.netty.bootstrap.AbstractBootstrap.initAndRegister(AbstractBootstrap.java:311)
    at io.netty.bootstrap.Bootstrap.doResolveAndConnect(Bootstrap.java:157)
    at io.netty.bootstrap.Bootstrap.connect(Bootstrap.java:141)
    at io.lettuce.core.AbstractRedisClient.initializeChannelAsync0(AbstractRedisClient.java:313)
    at io.lettuce.core.AbstractRedisClient.lambda$initializeChannelAsync$0(AbstractRedisClient.java:294)
    at reactor.core.publisher.LambdaMonoSubscriber.onNext(LambdaMonoSubscriber.java:168)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.lambda$onNext$0(TracingSubscriber.java:58)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.withActiveSpan(TracingSubscriber.java:79)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.onNext(TracingSubscriber.java:58)
    at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:203)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.lambda$onNext$0(TracingSubscriber.java:58)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.withActiveSpan(TracingSubscriber.java:79)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.onNext(TracingSubscriber.java:58)
    at reactor.core.publisher.FluxHide$SuppressFuseableSubscriber.onNext(FluxHide.java:127)
    at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onNext(MonoPeekTerminal.java:173)
    at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2148)
    at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.request(MonoPeekTerminal.java:132)
    at reactor.core.publisher.FluxHide$SuppressFuseableSubscriber.request(FluxHide.java:142)
    at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.request(FluxPeekFuseable.java:137)
    at reactor.core.publisher.LambdaMonoSubscriber.onSubscribe(LambdaMonoSubscriber.java:118)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.onSubscribe(TracingSubscriber.java:53)
    at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onSubscribe(FluxPeekFuseable.java:171)
    at io.opentelemetry.javaagent.shaded.instrumentation.reactor.TracingSubscriber.onSubscribe(TracingSubscriber.java:53)
    at reactor.core.publisher.FluxHide$SuppressFuseableSubscriber.onSubscribe(FluxHide.java:113)
    at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onSubscribe(MonoPeekTerminal.java:145)
    at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54)
    at reactor.core.publisher.Mono.subscribe(Mono.java:4087)
    at reactor.core.publisher.Mono.subscribeWith(Mono.java:4193)
    at reactor.core.publisher.Mono.subscribe(Mono.java:4059)
    at reactor.core.publisher.Mono.subscribe(Mono.java:3995)
    at reactor.core.publisher.Mono.subscribe(Mono.java:3967)
    at io.lettuce.core.AbstractRedisClient.initializeChannelAsync(AbstractRedisClient.java:289)
    at io.lettuce.core.cluster.RedisClusterClient.connectStatefulAsync(RedisClusterClient.java:705)
    at io.lettuce.core.cluster.RedisClusterClient.connectToNodeAsync(RedisClusterClient.java:499)
    at io.lettuce.core.cluster.RedisClusterClient$NodeConnectionFactoryImpl.connectToNodeAsync(RedisClusterClient.java:1171)
    at io.lettuce.core.cluster.topology.ClusterTopologyRefresh.getConnections(ClusterTopologyRefresh.java:263)
    at io.lettuce.core.cluster.topology.ClusterTopologyRefresh.loadViews(ClusterTopologyRefresh.java:74)
    at io.lettuce.core.cluster.RedisClusterClient.doLoadPartitions(RedisClusterClient.java:865)
    at io.lettuce.core.cluster.RedisClusterClient.loadPartitions(RedisClusterClient.java:845)
    at io.lettuce.core.cluster.RedisClusterClient.reloadPartitions(RedisClusterClient.java:784)
    at io.lettuce.core.cluster.ClusterTopologyRefreshScheduler$ClusterTopologyRefreshTask.doRun(ClusterTopologyRefreshScheduler.java:256)
    at io.lettuce.core.cluster.ClusterTopologyRefreshScheduler$ClusterTopologyRefreshTask.run(ClusterTopologyRefreshScheduler.java:236)
    at io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38)
    at io.netty.util.concurrent.PromiseTask.run(PromiseTask.java:73)
    at io.netty.util.concurrent.DefaultEventExecutor.run(DefaultEventExecutor.java:66)
    at io.netty.util.concurrent.SingleThreadEventExecutor$6.run(SingleThreadEventExecutor.java:1044)
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.lang.Thread.run(Thread.java:748)
2023-04-24 16:19:05.397 [lettuce-eventExecutorLoop-3-3] [ERROR] i.n.u.concurrent.DefaultPromise.rejectedExecution - Failed to submit a listener notification task. Event loop shut down?
java.util.concurrent.RejectedExecutionException: event executor terminated

@mp911de

[mp911de]

When destroying the connection factory, you no longer can submit tasks. If you need to swap out the connection factories, I suggest implementing the RedisConnectionFactory interface and delegating to an active LettuceConnectionFactory instead of destroying the top-level object.

[carresia]

ok, thanks a lot, I would try it.