public DLQEntry pollEntry(long timeout) throws IOException, InterruptedException {
        byte[] bytes = pollEntryBytes(timeout);
        if (bytes == null) {
            return null;
        }
        return DLQEntry.deserialize(bytes);
    }
@Test
    public void testReaderFindSegmentHoleAfterSimulatingRetentionPolicyClean() throws IOException, InterruptedException {
        final int eventsPerSegment = prepareFilledSegmentFiles(3);
        assertEquals(319, eventsPerSegment);

        int remainingEventsInSegment = eventsPerSegment;

        try (DeadLetterQueueReader reader = new DeadLetterQueueReader(dir)) {
            // read the first event to initialize reader structures
            final DLQEntry dlqEntry = reader.pollEntry(1_000);
            assertEquals("00000", dlqEntry.getReason());
            remainingEventsInSegment--;

            // simulate a storage policy clean, drop the middle segment file
            final List<Path> allSegments = listSegmentsSorted(dir);
            assertThat(allSegments.size(), greaterThanOrEqualTo(2));
            Files.delete(allSegments.remove(0)); // tail segment
            Files.delete(allSegments.remove(0)); // the segment after

            // consume the first segment
            for (int i = 0; i < remainingEventsInSegment; i++) {
                reader.pollEntry(1_000);
            }

            // Exercise
            // consume the first event after the hole
            final DLQEntry entryAfterHole = reader.pollEntry(1_000);

            assertEquals(String.format("%05d", eventsPerSegment * 2), entryAfterHole.getReason());
        }
    }