public void put(long cacheId, long position, ByteBuf data) {
        writeLock.lock();
        try {
            int dataLength = data.readableBytes();
            NavigableMap<Long, Blocks> cache = cacheMap.computeIfAbsent(cacheId, k -> new TreeMap<>());
            Map.Entry<Long, Blocks> pos2block = cache.floorEntry(position);
            long cacheStartPosition;
            long cacheEndPosition;
            Blocks blocks;
            boolean coverPosition = pos2block != null && position <= pos2block.getKey() + pos2block.getValue().dataLength;
            boolean tooManyBlocks = pos2block != null && pos2block.getValue().indexes.length >= Blocks.MAX_BLOCK_COUNT;
            if (!coverPosition || tooManyBlocks) {
                // no existing cache covers the position, or the existing cache has too many blocks, create a new one
                cacheStartPosition = position;
                cacheEndPosition = position + dataLength;
                blocks = Blocks.EMPTY;
            } else {
                // found an existing cache which covers the position, use it and maybe extend its capacity
                cacheStartPosition = pos2block.getKey();
                blocks = pos2block.getValue();
                cacheEndPosition = Math.max(pos2block.getKey() + blocks.dataLength, position + dataLength);
            }
            // ensure the capacity, if the capacity change then update the cache index
            int moreCapacity = (int) (cacheEndPosition - (cacheStartPosition + blocks.indexes.length * (long) blockSize));
            int newDataLength = (int) (cacheEndPosition - cacheStartPosition);
            if (moreCapacity > 0) {
                // need to extend the capacity, acquire free blocks
                int[] indexes = ensureCapacity(cacheStartPosition, moreCapacity);
                if (indexes == null) {
                    return;
                }
                int[] newIndexes = new int[blocks.indexes.length + indexes.length];
                System.arraycopy(blocks.indexes, 0, newIndexes, 0, blocks.indexes.length);
                System.arraycopy(indexes, 0, newIndexes, blocks.indexes.length, indexes.length);
                blocks = new Blocks(newIndexes, newDataLength);
            } else {
                blocks = new Blocks(blocks.indexes, newDataLength);
            }
            cache.put(cacheStartPosition, blocks);
            lru.put(new Key(cacheId, cacheStartPosition), blocks);

            // write data to cache
            ByteBuffer cacheByteBuffer = this.cacheByteBuffer.duplicate();
            int positionDelta = (int) (position - cacheStartPosition);
            int written = 0;
            ByteBuffer[] nioBuffers = data.nioBuffers();
            int[] indexes = blocks.indexes;
            for (ByteBuffer nioBuffer : nioBuffers) {
                ByteBuf buf = Unpooled.wrappedBuffer(nioBuffer);
                while (buf.readableBytes() > 0) {
                    int writePosition = positionDelta + written;
                    int index = indexes[writePosition / blockSize];
                    cacheByteBuffer.position(index * blockSize + writePosition % blockSize);
                    int length = Math.min(buf.readableBytes(), blockSize - writePosition % blockSize);
                    cacheByteBuffer.put(buf.slice(buf.readerIndex(), length).nioBuffer());
                    buf.skipBytes(length);
                    written += length;
                }
            }
        } finally {
            writeLock.unlock();
        }
    }
The index of the free block which will be checked next.
Used for acquiring free blocks.
core/src/main/scala/kafka/log/streamaspect/cache/FileCache.java