static Iterator<Map<Integer, List<Data>>> toBatches(final Iterator<Entry<Integer, Data>> entries,
                                                        final int maxBatch, Semaphore nodeWideLoadedKeyLimiter) {
        return new UnmodifiableIterator<>() {
            @Override
            public boolean hasNext() {
                return entries.hasNext();
            }

            @Override
            public Map<Integer, List<Data>> next() {
                if (!entries.hasNext()) {
                    throw new NoSuchElementException();
                }
                return nextBatch(entries, maxBatch, nodeWideLoadedKeyLimiter);
            }
        };
    }
@Test
    public void test_toBatches_with_nodeWideLimit() {
        int nodeWideLimit = 7;
        int entryCount = 100;

        Semaphore nodeWideLoadedKeyLimiter = new Semaphore(nodeWideLimit);
        Iterator<Map<Integer, List<Data>>> batches
                = MapKeyLoaderUtil.toBatches(newIterator(entryCount),
                1000, nodeWideLoadedKeyLimiter);

        List<Integer> batchSizes = new ArrayList<>();
        while (batches.hasNext()) {
            Map<Integer, List<Data>> batch = batches.next();
            batchSizes.add(batch.size());
            nodeWideLoadedKeyLimiter.release(batch.values().size());
        }

        // assert no batch is bigger than nodeWideLimit.
        assertEquals(0,
                batchSizes.stream().filter(size -> size > nodeWideLimit).count());
        // assert sum of all batches equals to entryCount.
        assertEquals(entryCount,
                batchSizes.stream().mapToInt(size -> size).sum());

    }