presto-hive/src/main/java/io/prestosql/plugin/hive/util/ThrottledAsyncQueue.java/ThrottledAsyncQueue/borrowBatchAsync
class ThrottledAsyncQueue:    @Override
    public synchronized <O> ListenableFuture<O> borrowBatchAsync(int maxSize, Function<List<T>, BorrowResult<T, O>> function)
    {
        checkArgument(maxSize >= 0, "maxSize must be at least 0");

        ListenableFuture<?> throttleFuture = immediateFuture(null);
        if (size() > 0) {
            // the queue is not empty, try to return a batch immediately if we are not throttled
            int size = maxBatchSize(maxSize);
            if (rateLimiter.tryAcquire(size)) {
                return super.borrowBatchAsync(size, function);
            }
        }
        else if (!isFinished()) {
            // the queue is empty but not finished, wait before we can query a batch
            throttleFuture = getNotEmptySignal();
        }

        return Futures.transformAsync(
                throttleFuture,
                any -> {
                    int size = maxBatchSize(maxSize);
                    if (size > 0) {
                        rateLimiter.acquire(size);
                    }
                    return super.borrowBatchAsync(size, function);
                },
                executor);
    }