docs: private <T> void executeOnPooledConnection(
        Address address,
        String type,
        Function<ClientConnection, CompletableFuture<T>> callback,
        Executor executor,
        CompletableFuture<T> future) {
      if (address.equals(returnAddress)) {
        callback.apply(localConnection).whenComplete((result, error) -> {
          if (error == null) {
            executor.execute(() -> future.complete(result));
          } else {
            executor.execute(() -> future.completeExceptionally(error));
          }
        });
        return;
      }
  
      channelPool.getChannel(address, type).whenComplete((channel, channelError) -> {
        if (channelError == null) {
          final ClientConnection connection = getOrCreateClientConnection(channel);
          callback.apply(connection).whenComplete((result, sendError) -> {
            if (sendError == null) {
              executor.execute(() -> future.complete(result));
            } else {
              final Throwable cause = Throwables.getRootCause(sendError);
              if (!(cause instanceof TimeoutException) && !(cause instanceof MessagingException)) {
                channel.close().addListener(f -> {
                  log.debug("Closing connection to {}", channel.remoteAddress());
                  connection.close();
                  connections.remove(channel);
                });
              }
              executor.execute(() -> future.completeExceptionally(sendError));
            }
          });
        } else {
          executor.execute(() -> future.completeExceptionally(channelError));
        }
      });
    }