private TimestampOffset readLargestTimestamp() throws IOException {
        // Get the last time index entry. If the time index is empty, it will return (-1, baseOffset)
        TimestampOffset lastTimeIndexEntry = timeIndex().lastEntry();
        OffsetPosition offsetPosition = offsetIndex().lookup(lastTimeIndexEntry.offset);

        // Scan the rest of the messages to see if there is a larger timestamp after the last time index entry.
        FileRecords.TimestampAndOffset maxTimestampOffsetAfterLastEntry = log.largestTimestampAfter(offsetPosition.position);
        if (maxTimestampOffsetAfterLastEntry.timestamp > lastTimeIndexEntry.timestamp)
            return new TimestampOffset(maxTimestampOffsetAfterLastEntry.timestamp, maxTimestampOffsetAfterLastEntry.offset);

        return lastTimeIndexEntry;
    }
Truncate off all index and log entries with offsets >= the given offset.
If the given offset is larger than the largest message in this segment, do nothing.

This method is not thread-safe.

@param offset The offset to truncate to
@return The number of log bytes truncated
storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java