StatefulRedisClusterPubSubConnectionImpl's activated() method will report exception after resubscribe() was call.

[JunRzz]

Bug Report

async.clusterMyId().thenAccept(this::setNodeId); in StatefulRedisClusterPubSubConnectionImpl's activated() will failed after
resubscribe(),and the exception was Command CLUSTER not allowed while subscribed. Allowed commands are: [PSUBSCRIBE, QUIT, PUNSUBSCRIBE, SUBSCRIBE, UNSUBSCRIBE, PING]

Current Behavior

I have some codes like this:

 ArrayList<RedisURI> list = new ArrayList<>();
        CharSequence password = "6lFaWttbWlpAZxlPoB8h";
        RedisURI redisUri = RedisURI.create("10.97.213.53", 6379);
        redisUri.setClientName("lettuce-test");
        StaticCredentialsProvider credentialsProvider = new StaticCredentialsProvider(null,
                password.toString().toCharArray());
        redisUri.setCredentialsProvider(credentialsProvider);
        list.add(redisUri);
        RedisClusterClient redisClusterClient = RedisClusterClient.create(list);
        redisClusterClient.addListener(new CommandListener() {
            @Override
            public void commandFailed(CommandFailedEvent event) {
                System.out.printf("Redis command %s failed:%s,",event.getCommand(),event.getCause().getMessage());
            }
        });
        StatefulRedisClusterPubSubConnection<String, String> connection = redisClusterClient.connectPubSub();
        connection.async().subscribe("test");
        connection.addListener(new RedisClusterPubSubListener<String, String>() {
              ....
        });

When I use client kill command to kill this sub connection on redis server,lettuce will reconnecting,and the CommandListener will print error message of cluster myid.

Environment

• Lettuce version(s): 6.2.1.RELEASE
• Redis version: 5.0.7

Possible Solution

StatefulRedisClusterPubSubConnectionImpl's resubscribe() set the nodeId at the same time.

@Override
    protected List<RedisFuture<Void>> resubscribe() {

        async().clusterMyId().thenAccept(nodeId -> {
            endpoint.setClusterNode(partitions.getPartitionByNodeId(nodeId));
            setNodeId(nodeId);
        });
        return super.resubscribe();
    }`

StatefulRedisClusterPubSubConnectionImpl's activated() add some check codes

  @Override
    public void activated() {
        super.activated();
        if (!endpoint.isSubscribed()){
            async.clusterMyId().thenAccept(this::setNodeId);
        }
    }

[mp911de]

That's fixed now.