Cannot refresh Redis Cluster topology

[jay-june]

Bug Report

Current Behavior

I've found that the WARN log message is "Cannot refresh Redis Cluster topology", after AWS Elasticache instance reboot.
Although the log was printed every topology refresh period, all requests was fine.

Stack trace

java.util.concurrent.CompletionException: java.lang.ClassCastException: class io.lettuce.core.protocol.CommandExpiryWriter cannot be cast to class io.lettuce.core.protocol.DefaultEndpoint (io.lettuce.core.protocol.CommandExpiryWriter and io.lettuce.core.protocol.DefaultEndpoint are in unnamed module of loader org.springframework.boot.loader.LaunchedURLClassLoader @41906a77)
	at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:314)
	at java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:319)
	at java.base/java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:718)
	at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
	at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073)
	at io.lettuce.core.cluster.RedisClusterClient.lambda$loadPartitionsAsync$34(RedisClusterClient.java:910)
	at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
	at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)
	at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
	at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073)
	at io.lettuce.core.RedisChannelHandler.lambda$closeAsync$0(RedisChannelHandler.java:154)
	at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
	at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)
	at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
	at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073)
	at io.lettuce.core.internal.Futures.lambda$adapt$1(Futures.java:120)
	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:984)
	at io.netty.channel.AbstractChannel$AbstractUnsafe.doClose0(AbstractChannel.java:761)
	at io.netty.channel.AbstractChannel$AbstractUnsafe.close(AbstractChannel.java:736)
	at io.netty.channel.AbstractChannel$AbstractUnsafe.close(AbstractChannel.java:607)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.close(DefaultChannelPipeline.java:1352)
	at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:620)
	at io.netty.channel.AbstractChannelHandlerContext.access$1200(AbstractChannelHandlerContext.java:59)
	at io.netty.channel.AbstractChannelHandlerContext$11.run(AbstractChannelHandlerContext.java:609)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
	at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:384)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
	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:834)
Caused by: java.lang.ClassCastException: class io.lettuce.core.protocol.CommandExpiryWriter cannot be cast to class io.lettuce.core.protocol.DefaultEndpoint (io.lettuce.core.protocol.CommandExpiryWriter and io.lettuce.core.protocol.DefaultEndpoint are in unnamed module of loader org.springframework.boot.loader.LaunchedURLClassLoader @41906a77)
	at io.lettuce.core.cluster.ClusterDistributionChannelWriter.disconnectDefaultEndpoint(ClusterDistributionChannelWriter.java:384)
	at io.lettuce.core.cluster.StatefulRedisClusterConnectionImpl.setPartitions(StatefulRedisClusterConnectionImpl.java:277)
	at io.lettuce.core.cluster.RedisClusterClient.lambda$updatePartitionsInConnections$29(RedisClusterClient.java:853)
	at io.lettuce.core.cluster.RedisClusterClient.forEachCloseable(RedisClusterClient.java:1088)
	at io.lettuce.core.cluster.RedisClusterClient.forEachClusterConnection(RedisClusterClient.java:1066)
	at io.lettuce.core.cluster.RedisClusterClient.updatePartitionsInConnections(RedisClusterClient.java:852)
	at io.lettuce.core.cluster.RedisClusterClient.lambda$refreshPartitionsAsync$28(RedisClusterClient.java:846)
	at java.base/java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:714)
	... 35 common frames omitted

Input Code

Input Code

public RedisClusterClient createClusterClient(boolean autoReconnect) {
    RedisClusterClient clusterClient = RedisClusterClient.create(this.resources, this.redisClusterProperty.getRedisURIs());

    ClusterTopologyRefreshOptions topologyRefreshOptions;

    topologyRefreshOptions = ClusterTopologyRefreshOptions.builder()
        .enablePeriodicRefresh(Duration.ofSeconds(10))
        .enableAllAdaptiveRefreshTriggers()
        .build();

    clusterClient.setOptions(
        ClusterClientOptions.builder()
            .autoReconnect(autoReconnect)
            .topologyRefreshOptions(topologyRefreshOptions)
            .build());
    return clusterClient;
  }

Expected behavior/code

"Cannot refresh Redis Cluster topology" log should no longer be printed.

Environment

• Lettuce version(s): 6.0.3.RELEASE
• Redis version: 5.0.6

[andreassalomonsson]

I'm seeing almost exactly the same issue.

Lettuce version: 6.1.0.RELEASE
ElastiCache Engine Compatability version: 6.0.5

Stack trace

java.util.concurrent.CompletionException: java.lang.ClassCastException: class io.lettuce.core.CommandListenerWriter cannot be cast to class io.lettuce.core.protocol.DefaultEndpoint (io.lettuce.core.CommandListenerWriter and io.lettuce.core.protocol.DefaultEndpoint are in unnamed module of loader 'app')
    at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(Unknown Source)
    at java.base/java.util.concurrent.CompletableFuture.completeThrowable(Unknown Source)
    at java.base/java.util.concurrent.CompletableFuture$UniAccept.tryFire(Unknown Source)
    at java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source)
    at java.base/java.util.concurrent.CompletableFuture.complete(Unknown Source)
    at io.lettuce.core.cluster.RedisClusterClient.lambda$loadPartitionsAsync$28(RedisClusterClient.java:927)
    at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(Unknown Source)
    at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(Unknown Source)
    at java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source)
    at java.base/java.util.concurrent.CompletableFuture.complete(Unknown Source)
    at io.lettuce.core.RedisChannelHandler.lambda$closeAsync$0(RedisChannelHandler.java:162)
    at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(Unknown Source)
    at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(Unknown Source)
    at java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source)
    at java.base/java.util.concurrent.CompletableFuture.complete(Unknown Source)
    at io.lettuce.core.internal.Futures.lambda$adapt$1(Futures.java:120)
    at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:578)
    at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:552)
    at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:491)
    at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:616)
    at io.netty.util.concurrent.DefaultPromise.setSuccess0(DefaultPromise.java:605)
    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:1000)
    at io.netty.channel.AbstractChannel$AbstractUnsafe.doClose0(AbstractChannel.java:762)
    at io.netty.channel.AbstractChannel$AbstractUnsafe.close(AbstractChannel.java:737)
    at io.netty.channel.AbstractChannel$AbstractUnsafe.close(AbstractChannel.java:608)
    at io.netty.channel.DefaultChannelPipeline$HeadContext.close(DefaultChannelPipeline.java:1352)
    at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:622)
    at io.netty.channel.AbstractChannelHandlerContext.access$1200(AbstractChannelHandlerContext.java:61)
    at io.netty.channel.AbstractChannelHandlerContext$11.run(AbstractChannelHandlerContext.java:611)
    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)
    at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
    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(Unknown Source)
Caused by: java.lang.ClassCastException: class io.lettuce.core.CommandListenerWriter cannot be cast to class io.lettuce.core.protocol.DefaultEndpoint (io.lettuce.core.CommandListenerWriter and io.lettuce.core.protocol.DefaultEndpoint are in unnamed module of loader 'app')
    at io.lettuce.core.cluster.ClusterDistributionChannelWriter.disconnectDefaultEndpoint(ClusterDistributionChannelWriter.java:428)
    at io.lettuce.core.cluster.StatefulRedisClusterConnectionImpl.setPartitions(StatefulRedisClusterConnectionImpl.java:272)
    at io.lettuce.core.cluster.RedisClusterClient.lambda$updatePartitionsInConnections$23(RedisClusterClient.java:870)
    at io.lettuce.core.cluster.RedisClusterClient.forEachCloseable(RedisClusterClient.java:1105)
    at io.lettuce.core.cluster.RedisClusterClient.forEachClusterConnection(RedisClusterClient.java:1083)
    at io.lettuce.core.cluster.RedisClusterClient.updatePartitionsInConnections(RedisClusterClient.java:869)
    at io.lettuce.core.cluster.RedisClusterClient.lambda$refreshPartitionsAsync$21(RedisClusterClient.java:863)
    ... 36 more

Code

    public RedisClusterClient createClusteredClient(RedisConfig redisConfig) {
        RedisClusterClient client = RedisClusterClient.create(clientResources, redisConfig.redisURI());

        ClusterTopologyRefreshOptions topologyRefreshOptions = ClusterTopologyRefreshOptions.builder()
                .enablePeriodicRefresh(Duration.of(10, ChronoUnit.MINUTES))
                .enableAllAdaptiveRefreshTriggers()
                .build();

        client.setOptions(ClusterClientOptions.builder()
                .topologyRefreshOptions(topologyRefreshOptions)
                .autoReconnect(redisConfig.isAutoReconnect())
                .pingBeforeActivateConnection(redisConfig.isPingBeforeActivateConnection())
                .cancelCommandsOnReconnectFailure(redisConfig.isCancelCommandsOnReconnetFailure())
                .socketOptions(SocketOptions.builder()
                        .keepAlive(redisConfig.isSoKeepAlive())
                        .connectTimeout(Duration.of(redisConfig.getSoConnectTimeoutMillis(), ChronoUnit.MILLIS))
                        .build())
                .build());

        return client;
    }