private int[] ensureCapacity(long cacheStartPosition, int size) {
        if (size > this.maxSize) {
            return null;
        }
        int requiredBlockCount = align(size) / blockSize;
        int[] indexes = new int[requiredBlockCount];
        int acquiringBlockIndex = 0;
        // evict the least recently used cache blocks until the free blocks are enough
        while (freeBlockCount + acquiringBlockIndex < requiredBlockCount) {
            Map.Entry<Key, Blocks> entry = lru.pop();
            if (entry == null) {
                break;
            }
            Key key = entry.getKey();
            Blocks blocks = entry.getValue();
            cacheMap.get(key.cacheId).remove(key.position);
            if (key.position == cacheStartPosition) {
                // eviction is conflict to current cache
                for (int i = 0; i < acquiringBlockIndex; i++) {
                    freeBlockCount++;
                    freeBlocks.set(indexes[i], true);
                }
                return null;
            }
            for (int blockIndex : blocks.indexes) {
                if (acquiringBlockIndex < indexes.length) {
                    indexes[acquiringBlockIndex++] = blockIndex;
                    freeBlocks.set(blockIndex, false);
                } else {
                    freeBlockCount++;
                    freeBlocks.set(blockIndex, true);
                }
            }
        }
        // acquire free blocks
        while (acquiringBlockIndex < indexes.length) {
            int next = freeBlocks.nextSetBit(freeCheckPoint);
            if (next >= 0) {
                indexes[acquiringBlockIndex++] = next;
                freeBlockCount--;
                freeBlocks.set(next, false);
                freeCheckPoint = next;
            } else if (freeCheckPoint != 0) {
                freeCheckPoint = 0;
            } else {
                // BUG
                return null;
            }
        }
        return indexes;
    }
Ensure the capacity of cache

@param cacheStartPosition if the eviction entries contain the current cache, then ensure capacity will return null.
@param size               size of data
@return the indexes of cache blocks
core/src/main/scala/kafka/log/streamaspect/cache/FileCache.java