presto-main/src/main/java/io/prestosql/operator/PartitionedConsumption.java/Partition/Partition
class Partition:        public Partition(
                int consumersCount,
                int partitionNumber,
                IntFunction<ListenableFuture<T>> loader,
                ListenableFuture<?> previousReleased,
                IntConsumer disposer)
        {
            this.partitionNumber = partitionNumber;
            this.requested = SettableFuture.create();
            this.loaded = Futures.transformAsync(
                    allAsList(requested, previousReleased),
                    ignored -> loader.apply(partitionNumber),
                    directExecutor());
            this.released = SettableFuture.create();
            released.addListener(() -> disposer.accept(partitionNumber), directExecutor());
            this.pendingReleases = consumersCount;
        }