public List<JournalReadEntry> readNext(long startOffset, long requestedMaximumCount) {
        // Capture the log end offset early for the failure handling below. The end offset will change during the
        // runtime of the retry loop because new messages are written to the journal. If we would use the changing
        // end offset in the error handling while loop, we would skip valid messages.
        final long logEndOffset = getLogEndOffset();

        List<JournalReadEntry> messages = read(startOffset, requestedMaximumCount);

        if (messages.isEmpty()) {
            // If we got an empty result BUT we know that there are more messages in the log, we bump the readOffset
            // by 1 and try to read again. We continue until we either get an non-empty result or we reached the
            // end of the log.
            // This can happen when a log segment is truncated at the end but later segments have valid messages again.
            long failedReadOffset = startOffset;
            long retryReadOffset = failedReadOffset + 1;

            while (messages.isEmpty() && failedReadOffset < (logEndOffset - 1)) {
                LOG.warn(
                        "Couldn't read any messages from offset <{}> but journal has more messages. Skipping and " +
                                "trying to read from offset <{}>",
                        failedReadOffset, retryReadOffset);

                // Retry the read with an increased offset to skip corrupt segments
                messages = read(retryReadOffset, requestedMaximumCount);

                // Bump offsets in case we still read an empty result
                failedReadOffset++;
                retryReadOffset++;
            }
        }

        return messages;
    }
@Test
    @Ignore
    public void readNext() throws Exception {
        final LocalKafkaJournal journal = new LocalKafkaJournal(journalDirectory.toPath(),
                scheduler, Size.kilobytes(1L),
                Duration.standardHours(1),
                Size.kilobytes(10L),
                Duration.standardDays(1),
                1_000_000,
                Duration.standardMinutes(1),
                100,
                new MetricRegistry(),
                serverStatus);

        final byte[] idBytes = "id".getBytes(UTF_8);
        final byte[] messageBytes = "message".getBytes(UTF_8);

         // High frequency reading with occasional writing.
        // This doesn't trigger the bug reliably every time though.
        IntStream.range(0, 1_000_000).parallel().forEach(
                (i) -> {
                    if (i % 1000 == 0) {
                        journal.write(idBytes, messageBytes);
                    } else {
                        journal.read(1L).forEach(e -> journal.markJournalOffsetCommitted(e.getOffset()));
                    }
                }
        );
    }