@Override
    public LogFetchInfo read(long startOffset, Isolation isolation) {
        verifyOffsetInRange(startOffset);

        long maxOffset = isolation == Isolation.COMMITTED ? highWatermark.offset() : endOffset().offset();
        if (startOffset >= maxOffset) {
            return new LogFetchInfo(MemoryRecords.EMPTY, new LogOffsetMetadata(
                startOffset, metadataForOffset(startOffset)));
        }

        ByteBuffer buffer = ByteBuffer.allocate(512);
        int batchCount = 0;
        LogOffsetMetadata batchStartOffset = null;

        logger.debug(
            "Looking for a batch that starts at {} and ends at {} for isolation {}",
            startOffset,
            maxOffset,
            isolation
        );

        for (LogBatch batch : batches) {
            // Note that start offset is inclusive while max offset is exclusive. We only return
            // complete batches, so batches which end at an offset larger than the max offset are
            // filtered, which is effectively the same as having the consumer drop an incomplete
            // batch returned in a fetch response.
            if (batch.lastOffset() >= startOffset && batch.lastOffset() < maxOffset && !batch.entries.isEmpty()) {
                buffer = batch.writeTo(buffer);

                if (batchStartOffset == null) {
                    batchStartOffset = batch.entries.get(0).logOffsetMetadata();
                }

                // Read on the mock log should return at most 2 batches. This is a simple solution
                // for testing interesting partial read scenarios.
                batchCount += 1;
                if (batchCount >= 2) {
                    break;
                }
            }
        }

        buffer.flip();
        Records records = MemoryRecords.readableRecords(buffer);

        if (batchStartOffset == null) {
            throw new RuntimeException("Expected to find at least one entry starting from offset " +
                startOffset + " but found none");
        }

        return new LogFetchInfo(records, batchStartOffset);
    }
Reopening the log causes all unflushed data to be lost.
raft/src/test/java/org/apache/kafka/raft/MockLog.java