docs: private <K, V> CompletableFuture<StatefulRedisClusterPubSubConnection<K, V>> connectClusterPubSubAsync(
              RedisCodec<K, V> codec) {
  
          if (partitions == null) {
              return Futures.failed(new IllegalStateException(
                      "Partitions not initialized. Initialize via RedisClusterClient.getPartitions()."));
          }
  
          activateTopologyRefreshIfNeeded();
  
          logger.debug("connectClusterPubSub(" + initialUris + ")");
  
          Mono<SocketAddress> socketAddressSupplier = getSocketAddressSupplier(TopologyComparators::sortByClientCount);
  
          PubSubClusterEndpoint<K, V> endpoint = new PubSubClusterEndpoint<K, V>(clientOptions, clientResources);
          RedisChannelWriter writer = endpoint;
  
          if (CommandExpiryWriter.isSupported(clientOptions)) {
              writer = new CommandExpiryWriter(writer, clientOptions, clientResources);
          }
  
          ClusterDistributionChannelWriter clusterWriter = new ClusterDistributionChannelWriter(clientOptions, writer,
                  clusterTopologyRefreshScheduler);
  
          StatefulRedisClusterPubSubConnectionImpl<K, V> connection = new StatefulRedisClusterPubSubConnectionImpl<>(endpoint,
                  clusterWriter, codec, timeout);
  
          ClusterPubSubConnectionProvider<K, V> pooledClusterConnectionProvider = new ClusterPubSubConnectionProvider<>(this,
                  clusterWriter, codec, connection.getUpstreamListener(), clusterTopologyRefreshScheduler);
  
          clusterWriter.setClusterConnectionProvider(pooledClusterConnectionProvider);
          connection.setPartitions(partitions);
  
          Supplier<CommandHandler> commandHandlerSupplier = () -> new PubSubCommandHandler<>(clientOptions, clientResources,
                  codec, endpoint);
  
          Mono<StatefulRedisClusterPubSubConnectionImpl<K, V>> connectionMono = Mono.defer(() -> connect(socketAddressSupplier,
                  codec, endpoint, connection, commandHandlerSupplier));
  
          for (int i = 1; i < getConnectionAttempts(); i++) {
              connectionMono = connectionMono.onErrorResume(t -> connect(socketAddressSupplier, codec, endpoint, connection,
                      commandHandlerSupplier));
          }
  
          return connectionMono
                  .flatMap(c -> c.reactive().command().collectList() //
                          .map(CommandDetailParser::parse) //
                                  .doOnNext(detail -> c.setState(new RedisState(detail)))
                                  .doOnError(e -> c.setState(new RedisState(Collections.emptyList()))).then(Mono.just(c))
                                  .onErrorResume(RedisCommandExecutionException.class, e -> Mono.just(c)))
                  .doOnNext(
                          c -> connection.registerCloseables(closeableResources, clusterWriter, pooledClusterConnectionProvider))
                  .map(it -> (StatefulRedisClusterPubSubConnection<K, V>) it).toFuture();
      }