docs: <K, V> ConnectionFuture<StatefulRedisPubSubConnection<K, V>> connectPubSubToNodeAsync(RedisCodec<K, V> codec,
              String nodeId, Mono<SocketAddress> socketAddressSupplier) {
  
          assertNotNull(codec);
          assertNotEmpty(initialUris);
  
          LettuceAssert.notNull(socketAddressSupplier, "SocketAddressSupplier must not be null");
  
          logger.debug("connectPubSubToNode(" + nodeId + ")");
  
          PubSubEndpoint<K, V> endpoint = new PubSubEndpoint<>(clientOptions, clientResources);
  
          RedisChannelWriter writer = endpoint;
  
          if (CommandExpiryWriter.isSupported(clientOptions)) {
              writer = new CommandExpiryWriter(writer, clientOptions, clientResources);
          }
  
          StatefulRedisPubSubConnectionImpl<K, V> connection = new StatefulRedisPubSubConnectionImpl<>(endpoint, writer, codec,
                  timeout);
  
          ConnectionFuture<StatefulRedisPubSubConnection<K, V>> connectionFuture = connectStatefulAsync(connection, codec,
                  endpoint,
                  getFirstUri(), socketAddressSupplier, () -> new PubSubCommandHandler<>(clientOptions, clientResources, codec,
                          endpoint));
          return connectionFuture.whenComplete((conn, throwable) -> {
              if (throwable != null) {
                  connection.close();
              }
          });
      }