@Test
public void messagesArePassedToInnerWriter() throws IOException {

    final CloseableWriter<Integer> innerMock = mock(CloseableWriter.class);
    final ExpiringConsumer<Integer> consumer = new ExpiringConsumer<>(innerMock, Duration.ZERO, 5);
    final List<Offset> offsets = Arrays.asList(buildOffset(11), buildOffset(12), buildOffset(13));
    for (int i = 0; i < 3; i++) {
        consumer.write(1234567890L, i, offsets.get(i));
    }
    final InOrder inOrder = inOrder(innerMock);
    for (int i = 0; i < 3; i++) {
        inOrder.verify(innerMock).write(1234567890L, i, offsets.get(i));
    }
    verifyNoMoreInteractions(innerMock);
}