@Override
    public LogSegment add(LogSegment segment) {
        synchronized (this) {
            if (null == activeSegment || segment.baseOffset() >= activeSegment.baseOffset()) {
                activeSegment = segment;
            }
            return super.add(segment);
        }
    }
Its [[LogSegment.baseOffset]] will always be the largest one among all segments.
core/src/main/scala/kafka/log/streamaspect/CachedLogSegments.java