docs: <K, V> ConnectionFuture<StatefulRedisConnection<K, V>> connectToNodeAsync(RedisCodec<K, V> codec, String nodeId,
              RedisChannelWriter clusterWriter, Mono<SocketAddress> socketAddressSupplier) {
  
          assertNotNull(codec);
          assertNotEmpty(initialUris);
          LettuceAssert.notNull(socketAddressSupplier, "SocketAddressSupplier must not be null");
  
          ClusterNodeEndpoint endpoint = new ClusterNodeEndpoint(clientOptions, getResources(), clusterWriter);
  
          RedisChannelWriter writer = endpoint;
  
          if (CommandExpiryWriter.isSupported(clientOptions)) {
              writer = new CommandExpiryWriter(writer, clientOptions, clientResources);
          }
  
          StatefulRedisConnectionImpl<K, V> connection = new StatefulRedisConnectionImpl<K, V>(writer, codec, timeout);
  
          ConnectionFuture<StatefulRedisConnection<K, V>> connectionFuture = connectStatefulAsync(connection, codec, endpoint,
                  getFirstUri(), socketAddressSupplier, () -> new CommandHandler(clientOptions, clientResources, endpoint));
  
          return connectionFuture.whenComplete((conn, throwable) -> {
              if (throwable != null) {
                  connection.close();
              }
          });
      }