presto-main/src/main/java/io/prestosql/execution/scheduler/group/FixedLifespanScheduler.java/FixedLifespanScheduler/FixedLifespanScheduler
class FixedLifespanScheduler:    public FixedLifespanScheduler(BucketNodeMap bucketNodeMap, List<ConnectorPartitionHandle> partitionHandles, OptionalInt concurrentLifespansPerTask)
    {
        checkArgument(!partitionHandles.equals(ImmutableList.of(NOT_PARTITIONED)));
        checkArgument(partitionHandles.size() == bucketNodeMap.getBucketCount());

        Map<InternalNode, IntList> nodeToDriverGroupMap = new HashMap<>();
        Int2ObjectMap<InternalNode> driverGroupToNodeMap = new Int2ObjectOpenHashMap<>();
        for (int bucket = 0; bucket < bucketNodeMap.getBucketCount(); bucket++) {
            InternalNode node = bucketNodeMap.getAssignedNode(bucket).get();
            nodeToDriverGroupMap.computeIfAbsent(node, key -> new IntArrayList()).add(bucket);
            driverGroupToNodeMap.put(bucket, node);
        }

        this.driverGroupToNodeMap = driverGroupToNodeMap;
        this.nodeToDriverGroupsMap = nodeToDriverGroupMap.entrySet().stream()
                .collect(toImmutableMap(Map.Entry::getKey, entry -> entry.getValue().iterator()));
        this.partitionHandles = requireNonNull(partitionHandles, "partitionHandles is null");
        if (concurrentLifespansPerTask.isPresent()) {
            checkArgument(concurrentLifespansPerTask.getAsInt() >= 1, "concurrentLifespansPerTask must be great or equal to 1 if present");
        }
        this.concurrentLifespansPerTask = requireNonNull(concurrentLifespansPerTask, "concurrentLifespansPerTask is null");
    }