@Override
    public List<LogicalSlot> peakSlotsToAllocate(SlotTracker slotTracker) {
        updateOptionsPeriodically();

        List<LogicalSlot> slotsToAllocate = Lists.newArrayList();

        int curNumAllocatedSmallSlots = numAllocatedSmallSlots;
        for (SlotContext slotContext : requiringSmallSlots.values()) {
            LogicalSlot slot = slotContext.getSlot();
            if (curNumAllocatedSmallSlots + slot.getNumPhysicalSlots() > opts.v2().getTotalSmallSlots()) {
                break;
            }

            requiringQueue.remove(slotContext);

            slotsToAllocate.add(slot);
            slotContext.setAllocateAsSmallSlot();
            curNumAllocatedSmallSlots += slot.getNumPhysicalSlots();
        }

        int numAllocatedSlots = slotTracker.getNumAllocatedSlots() - numAllocatedSmallSlots;
        while (!requiringQueue.isEmpty()) {
            SlotContext slotContext = requiringQueue.peak();
            if (!isGlobalSlotAvailable(numAllocatedSlots, slotContext.getSlot())) {
                break;
            }

            requiringQueue.poll();

            slotsToAllocate.add(slotContext.getSlot());
            numAllocatedSlots += slotContext.getSlot().getNumPhysicalSlots();
        }

        return slotsToAllocate;
    }
@Test
    public void testHeadLineBlocking1() {
        QueryQueueOptions opts = QueryQueueOptions.createFromEnv();
        SlotSelectionStrategyV2 strategy = new SlotSelectionStrategyV2();
        SlotTracker slotTracker = new SlotTracker(ImmutableList.of(strategy));

        LogicalSlot slot1 = generateSlot(opts.v2().getTotalSlots() / 2 + 1);
        LogicalSlot slot2 = generateSlot(opts.v2().getTotalSlots() / 2);
        LogicalSlot slot3 = generateSlot(2);

        // 1. Require and allocate slot1.
        slotTracker.requireSlot(slot1);
        assertThat(strategy.peakSlotsToAllocate(slotTracker)).containsExactly(slot1);
        slotTracker.allocateSlot(slot1);

        // 2. Require slot2.
        slotTracker.requireSlot(slot2);
        assertThat(strategy.peakSlotsToAllocate(slotTracker)).isEmpty();

        // 3. Require enough small slots to make its priority lower.
        {
            List<LogicalSlot> smallSlots = IntStream.range(0, 10)
                    .mapToObj(i -> generateSlot(2))
                    .collect(Collectors.toList());
            smallSlots.forEach(slotTracker::requireSlot);
            for (int numPeakedSmallSlots = 0; numPeakedSmallSlots < 10; ) {
                List<LogicalSlot> peakSlots = strategy.peakSlotsToAllocate(slotTracker);
                numPeakedSmallSlots += peakSlots.size();
                peakSlots.forEach(slotTracker::allocateSlot);
                peakSlots.forEach(slot -> assertThat(slotTracker.releaseSlot(slot.getSlotId())).isSameAs(slot));
            }
        }

        // Try peak the only rest slot2, but it is blocked by slot1.
        assertThat(strategy.peakSlotsToAllocate(slotTracker)).isEmpty();

        // 4. slot3 cannot be peaked because it is blocked by slot2.
        slotTracker.requireSlot(slot3);
        assertThat(strategy.peakSlotsToAllocate(slotTracker)).isEmpty();

        // 5. slot3 can be peaked after releasing the pending slot2.
        assertThat(slotTracker.releaseSlot(slot2.getSlotId())).isSameAs(slot2);
        assertThat(strategy.peakSlotsToAllocate(slotTracker)).containsExactly(slot3);
        slotTracker.allocateSlot(slot3);
        assertThat(slotTracker.releaseSlot(slot3.getSlotId())).isSameAs(slot3);
    }