docs: private <K, V, T extends RedisChannelHandler<K, V>, S> ConnectionFuture<S> connectStatefulAsync(T connection,
              RedisCodec<K, V> codec,
              DefaultEndpoint endpoint, RedisURI connectionSettings, Mono<SocketAddress> socketAddressSupplier,
              Supplier<CommandHandler> commandHandlerSupplier) {
  
          ConnectionBuilder connectionBuilder = createConnectionBuilder(connection, endpoint, connectionSettings,
                  socketAddressSupplier, commandHandlerSupplier);
  
          if (clientOptions.isPingBeforeActivateConnection()) {
              if (hasPassword(connectionSettings)) {
                  connectionBuilder.enableAuthPingBeforeConnect();
              } else {
                  connectionBuilder.enablePingBeforeConnect();
              }
          }
  
          ConnectionFuture<RedisChannelHandler<K, V>> future = initializeChannelAsync(connectionBuilder);
          ConnectionFuture<?> sync = future;
  
          if (!clientOptions.isPingBeforeActivateConnection() && hasPassword(connectionSettings)) {
  
              sync = sync.thenCompose(channelHandler -> {
  
                  CommandArgs<K, V> args = new CommandArgs<>(codec).add(connectionSettings.getPassword());
                  AsyncCommand<K, V, String> command = new AsyncCommand<>(new Command<>(CommandType.AUTH, new StatusOutput<>(
                          codec), args));
  
                  if (connection instanceof StatefulRedisClusterConnectionImpl) {
                      ((StatefulRedisClusterConnectionImpl) connection).dispatch(command);
                  }
  
                  if (connection instanceof StatefulRedisConnectionImpl) {
                      ((StatefulRedisConnectionImpl) connection).dispatch(command);
                  }
  
                  return command;
              });
  
          }
  
          if (LettuceStrings.isNotEmpty(connectionSettings.getClientName())) {
              sync = sync.thenApply(channelHandler -> {
  
                  if (connection instanceof StatefulRedisClusterConnectionImpl) {
                      ((StatefulRedisClusterConnectionImpl) connection).setClientName(connectionSettings.getClientName());
                  }
  
                  if (connection instanceof StatefulRedisConnectionImpl) {
                      ((StatefulRedisConnectionImpl) connection).setClientName(connectionSettings.getClientName());
                  }
                  return channelHandler;
              });
          }
  
          return sync.thenApply(channelHandler -> (S) connection);
      }