@Override
    public List<JournalReadEntry> read(long requestedMaximumCount) {
        return readNext(nextReadOffset, requestedMaximumCount);
    }
@Test
    public void truncatedSegment() throws Exception {
        final Size segmentSize = Size.kilobytes(1L);
        final LocalKafkaJournal journal = new LocalKafkaJournal(journalDirectory.toPath(),
                scheduler,
                segmentSize,
                Duration.standardHours(1),
                Size.kilobytes(10L),
                Duration.standardDays(1),
                1_000_000,
                Duration.standardMinutes(1),
                100,
                new MetricRegistry(),
                serverStatus);

        // this will create two segments, each containing 25 messages
        createBulkChunks(journal, segmentSize, 2);

        final Path firstSegmentPath =
                Paths.get(journalDirectory.getAbsolutePath(), "messagejournal-0", "00000000000000000000.log");

        assertThat(firstSegmentPath).isRegularFile();

        // truncate the first segment so that the last message is cut off
        final File firstSegment = firstSegmentPath.toFile();
        try (FileChannel channel = new FileOutputStream(firstSegment, true).getChannel()) {
            channel.truncate(firstSegment.length() - 1);
        }

        final List<Journal.JournalReadEntry> entriesFromFirstSegment = journal.read(25);
        assertThat(entriesFromFirstSegment).hasSize(24);
        final List<Journal.JournalReadEntry> entriesFromSecondSegment = journal.read(25);
        assertThat(entriesFromSecondSegment).hasSize(25);
    }