Kafka stores each partition as a sequence of log segment files -- plain append-only files of records. Finding a specific offset by scanning a whole segment from the start would be slow for a segment with millions of records, so each segment also has a SPARSE offset index: not every record gets an index entry (that would cost as much memory as the log itself), only one every index_interval_bytes bytes written. To read a target offset, the reader looks up the nearest indexed byte position AT OR BEFORE the target (the "floor" entry) and scans forward from there -- a handful of records instead of potentially the whole segment.

Implement:

• OffsetIndex.lookup(target_relative_offset) -- return the byte position of the indexed entry with the LARGEST relative_offset that is <= target_relative_offset. If no such entry exists (the index is empty, or every entry's relative_offset is greater than the target), return 0 (start of the segment).
• LogSegment.append(offset, payload) -- append (offset, payload) to the segment. Add a sparse index entry for THIS record (its relative_offset = offset - base_offset and its byte position) only if _bytes_since_last_index has already reached index_interval_bytes -- then reset that counter to 0. Either way, advance size and _bytes_since_last_index by len(payload).
• LogSegment.read(target_offset) -- use the index to find a starting byte position, then scan self.log forward from there for the record whose offset == target_offset. Return its payload, or None if not found.

Example

seg = LogSegment(base_offset=100, index_interval_bytes=9)
for offset in range(100, 111):
    seg.append(offset, str(offset).encode())  # 3-byte payloads

seg.index.entries   # -> [(3, 9), (6, 18), (9, 27)] -- sparse, not per-record
seg.read(105)        # -> b"105"  (found via the floor index entry, not a full scan)
seg.read(999)         # -> None

Constraints

• The index check in append() uses the counter's value BEFORE this record's own bytes are added -- a new entry, once added, always points at the position of the record that JUST crossed the threshold.
• lookup() never raises for an out-of-range target -- it returns 0 (or the last entry) instead.

Real-world grounding

• Kafka's log segment (storage moved from Scala to Java in recent releases): storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java
• Kafka's sparse offset index: OffsetIndex.java

Real Kafka indexes by BYTES written (as modeled here) and also maintains a separate time index (timestamp -> offset) on top of this one -- but "sparse index, floor lookup, then a short forward scan" is the core O(log n) trick underneath both.