private TimestampOffset parseEntry0(int n) throws ExecutionException, InterruptedException {
        int startOffset = n * ENTRY_SIZE;
        TimestampOffset timestampOffset = tryGetEntryFromCache(n);
        if (!TimestampOffset.UNKNOWN.equals(timestampOffset)) {
            return timestampOffset;
        }
        // cache missing, try to read from remote and put it to cache.
        // the index interval is 1MiB and the segment size is 1GB, so binary search only need 512 entries
        long endOffset = Math.min(entries() * ENTRY_SIZE, startOffset + ENTRY_SIZE * 512);
        if (endOffset > stream.confirmOffset()) {
            // the end offset is beyond the confirmed offset, so we need to wait for the last append to finish
            //  to ensure the data is available
            try {
                lastAppend.get();
            } catch (Throwable e) {
                throw new RuntimeException(e);
            }
        }
        FetchResult rst = stream.fetch(startOffset, endOffset).get();
        List<RecordBatchWithContext> records = rst.recordBatchList();
        if (records.isEmpty()) {
            throw new IllegalStateException("fetch empty from stream " + stream + " at offset " + startOffset);
        }
        ByteBuf buf = Unpooled.buffer(records.size() * ENTRY_SIZE);
        records.forEach(record -> buf.writeBytes(record.rawPayload()));
        cache.put(cacheId, startOffset, buf);
        ByteBuf indexEntry = Unpooled.wrappedBuffer(records.get(0).rawPayload());
        timestampOffset = new TimestampOffset(indexEntry.readLong(), baseOffset() + indexEntry.readInt());
        rst.free();
        return timestampOffset;
    }
In the case of unclean shutdown, the last entry needs to be recovered from the time index.
core/src/main/scala/kafka/log/streamaspect/ElasticTimeIndex.java