Repository: apache/kafka
File: storage/src/main/java/org/apache/kafka/storage/internals/log/OffsetIndex.java (Java)

Code before:
            log.debug("Loaded index file {} with maxEntries = {}, maxIndexSize = {}, entries = {}, lastOffset = {}, file position = {}",
                file.getAbsolutePath(), maxEntries(), maxIndexSize, entries(), lastOffset, mmap().position());
        });
    }

    @Override
    public void sanityCheck() {
        if (entries() != 0 && lastOffset < baseOffset())
            throw new CorruptIndexException("Corrupt index found, index file " + file().getAbsolutePath() + " has non-zero size " +
                "but the last offset is " + lastOffset + " which is less than the base offset " + baseOffset());
        if (length() % entrySize() != 0)
            throw new CorruptIndexException("Index file " + file().getAbsolutePath() + " is corrupt, found " + length() +
                " bytes which is neither positive nor a multiple of " + ENTRY_SIZE);
    }

    /**
     * Find the largest offset less than or equal to the given targetOffset
     * and return a pair holding this offset and its corresponding physical file position.
     *
     * @param targetOffset The offset to look up.
     * @return The offset found and the corresponding file position for this offset
     *         If the target offset is smaller than the least entry in the index (or the index is empty),
     *         the pair (baseOffset, 0) is returned.
     */
    public OffsetPosition lookup(long targetOffset) {
        return inLock(lock, () -> {
            ByteBuffer idx = mmap().duplicate();
            int slot = largestLowerBoundSlotFor(idx, targetOffset, IndexSearchType.KEY);
            if (slot == -1)
                return new OffsetPosition(baseOffset(), 0);
            else
                return parseEntry(idx, slot);
        });
    }

    /**
     * Get the nth offset mapping from the index
     * @param n The entry number in the index
     * @return The offset/position pair at that entry
     */
    public OffsetPosition entry(int n) {
        return inLock(lock, () -> {
            if (n >= entries())
                throw new IllegalArgumentException("Attempt to fetch the " + n + "th entry from index " +
                    file().getAbsolutePath() + ", which has size " + entries());
            return parseEntry(mmap(), n);
        });
    }

    /**
     * Find an upper bound offset for the given fetch starting position and size. This is an offset which

Review comment (suggestion):
Currently, in Linux, we don't acquire the lock for `lookup()`, which is used for fetch requests. Both fetch and produce requests are common. Acquiring the lock in the fetch path reduces read/write concurrency. The reader only truly needs to lock when the underlying mmap changes by resize(). Since resize() is an infrequent event, we could introduce a separate resize lock, which will be held by `resize()` and all readers (currently calling `maybeLock()`). This will help maintain the current level of concurrency. 

Code after:

    @Override
    public void sanityCheck() {
        if (entries() != 0 && lastOffset < baseOffset())
            throw new CorruptIndexException("Corrupt index found, index file " + file().getAbsolutePath() + " has non-zero size " +
                "but the last offset is " + lastOffset + " which is less than the base offset " + baseOffset());
        if (length() % entrySize() != 0)
            throw new CorruptIndexException("Index file " + file().getAbsolutePath() + " is corrupt, found " + length() +
                " bytes which is neither positive nor a multiple of " + ENTRY_SIZE);
    }

    /**
     * Find the largest offset less than or equal to the given targetOffset
     * and return a pair holding this offset and its corresponding physical file position.
     *
     * @param targetOffset The offset to look up.
     * @return The offset found and the corresponding file position for this offset
     *         If the target offset is smaller than the least entry in the index (or the index is empty),
     *         the pair (baseOffset, 0) is returned.
     */
    public OffsetPosition lookup(long targetOffset) {
        return inRemapReadLock(() -> {
            ByteBuffer idx = mmap().duplicate();
            int slot = largestLowerBoundSlotFor(idx, targetOffset, IndexSearchType.KEY);
            if (slot == -1)
                return new OffsetPosition(baseOffset(), 0);
            else
                return parseEntry(idx, slot);
        });
    }

    /**
     * Get the nth offset mapping from the index
     * @param n The entry number in the index
     * @return The offset/position pair at that entry
     */
    public OffsetPosition entry(int n) {
        return inRemapReadLock(() -> {
            if (n >= entries())
                throw new IllegalArgumentException("Attempt to fetch the " + n + "th entry from index " +
                    file().getAbsolutePath() + ", which has size " + entries());
            return parseEntry(mmap(), n);
        });
    }

    /**
     * Find an upper bound offset for the given fetch starting position and size. This is an offset which
     * is guaranteed to be outside the fetched range, but note that it will not generally be the smallest
     * such offset.
     */
    public Optional<OffsetPosition> fetchUpperBoundOffset(OffsetPosition fetchOffset, int fetchSize) {