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

Code before:

    public void resizeIndexes(int size) throws IOException {
        offsetIndex().resize(size);
        timeIndex().resize(size);
    }

    public void sanityCheck(boolean timeIndexFileNewlyCreated) throws IOException {
        if (offsetIndexFile().exists()) {
            // Resize the time index file to 0 if it is newly created.
            if (timeIndexFileNewlyCreated) 
                timeIndex().resize(0);
            // Sanity checks for time index and offset index are skipped because
            // we will recover the segments above the recovery point in recoverLog()
            // in any case so sanity checking them here is redundant.
            txnIndex.sanityCheck();
        } else
            throw new NoSuchFileException("Offset index file " + offsetIndexFile().getAbsolutePath() + " does not exist");
    }

    /**
     * The first time this is invoked, it will result in a time index lookup (including potential materialization of
     * the time index).
     */
    public TimestampOffset readMaxTimestampAndOffsetSoFar() throws IOException {
        if (maxTimestampAndOffsetSoFar == TimestampOffset.UNKNOWN) {
            synchronized (this) {
                if (maxTimestampAndOffsetSoFar == TimestampOffset.UNKNOWN) {
                    maxTimestampAndOffsetSoFar = timeIndex().lastEntry();
                }
            }
        }
        return maxTimestampAndOffsetSoFar;
    }

    /**
     * The maximum timestamp we see so far.
     *
     * Note that this may result in time index materialization.
     */
    public long maxTimestampSoFar() throws IOException {
        return readMaxTimestampAndOffsetSoFar().timestamp;
    }

    /**
     * Note that this may result in time index materialization.
     */
    private long shallowOffsetOfMaxTimestampSoFar() throws IOException {
        return readMaxTimestampAndOffsetSoFar().offset;
    }

    /* Return the size in bytes of this log segment */

Review comment (suggestion):
The monitor lock of `LogSegment` may be used by external code. Perhaps it would be safer to use the monitor lock of an internal `Object` instead?
```java
private final Object maxTimestampAndOffsetLock = new Object();

synchronized (maxTimestampAndOffsetLock) {
}

```

Code after:
            (size > 0 && reachedRollMs) ||
            offsetIndex().isFull() || timeIndex().isFull() || !canConvertToRelativeOffset(rollParams.maxOffsetInMessages);
    }

    public void resizeIndexes(int size) throws IOException {
        offsetIndex().resize(size);
        timeIndex().resize(size);
    }

    public void sanityCheck(boolean timeIndexFileNewlyCreated) throws IOException {
        if (offsetIndexFile().exists()) {
            // Resize the time index file to 0 if it is newly created.
            if (timeIndexFileNewlyCreated)
                timeIndex().resize(0);
            // Sanity checks for time index and offset index are skipped because
            // we will recover the segments above the recovery point in recoverLog()
            // in any case so sanity checking them here is redundant.
            txnIndex.sanityCheck();
        } else
            throw new NoSuchFileException("Offset index file " + offsetIndexFile().getAbsolutePath() + " does not exist");
    }

    /**
     * The first time this is invoked, it will result in a time index lookup (including potential materialization of
     * the time index).
     */
    public TimestampOffset readMaxTimestampAndOffsetSoFar() throws IOException {
        if (maxTimestampAndOffsetSoFar == TimestampOffset.UNKNOWN) {
            // As stated in LogSegment class javadoc, this class is not thread-safe so basically we assume that
            // methods are called within UnifiedLog#lock.
            // However, there's exceptional paths where this method can be called outside of the lock,
            // so we need lock here to prevent multiple threads trying to modify maxTimestampAndOffsetSoFar
            synchronized (maxTimestampAndOffsetLock) {
                if (maxTimestampAndOffsetSoFar == TimestampOffset.UNKNOWN) {
                    maxTimestampAndOffsetSoFar = timeIndex().lastEntry();
                }
            }
        }
        return maxTimestampAndOffsetSoFar;
    }

    /**
     * The maximum timestamp we see so far.
     *
     * Note that this may result in time index materialization.
     */
    public long maxTimestampSoFar() throws IOException {
        return readMaxTimestampAndOffsetSoFar().timestamp;
    }

    /**