@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 testUpdateOptionsPeriodicallyAtReleasing() throws InterruptedException {
        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 - 1);
        LogicalSlot slot3 = generateSlot(2);

        // 1. Require slot1, slot2, slot3.
        assertThat(slotTracker.requireSlot(slot1)).isTrue();
        assertThat(slotTracker.requireSlot(slot2)).isTrue();
        assertThat(slotTracker.requireSlot(slot3)).isTrue();

        // 2. Peak, allocate slot2 and slot3.
        assertThat(strategy.peakSlotsToAllocate(slotTracker)).containsExactly(slot3, slot2);
        slotTracker.allocateSlot(slot2);
        slotTracker.allocateSlot(slot3);
        assertThat(slot2.getState()).isEqualTo(LogicalSlot.State.ALLOCATED);
        assertThat(slot3.getState()).isEqualTo(LogicalSlot.State.ALLOCATED);
        assertThat(slotTracker.getNumAllocatedSlots()).isEqualTo(slot2.getNumPhysicalSlots() + slot3.getNumPhysicalSlots());

        // Make options changed.
        BackendResourceStat.getInstance().setNumHardwareCoresOfBe(1, NUM_CORES * 2);
        Thread.sleep(1200);
        // 3. Release slot2 and slot3.
        assertThat(slotTracker.releaseSlot(slot2.getSlotId())).isSameAs(slot2);
        assertThat(slotTracker.releaseSlot(slot3.getSlotId())).isSameAs(slot3);

        // 4. Peak slot1.
        assertThat(strategy.peakSlotsToAllocate(slotTracker)).containsExactly(slot1);
        slotTracker.allocateSlot(slot1);
        assertThat(slot1.getState()).isEqualTo(LogicalSlot.State.ALLOCATED);
        assertThat(slotTracker.getNumAllocatedSlots()).isEqualTo(slot1.getNumPhysicalSlots());
    }