ExpiringConsumer implements CloseableWriter<MESSAGE_TYPE> { @Override public void write(long timestamp, MESSAGE_TYPE message, Offset offset) throws IOException { ++this.messagesReceived; this.writer.write(timestamp, message, offset); } ExpiringConsumer(CloseableWriter<MESSAGE_TYPE> writer, TemporalAmount expirationDelay, long messagesBeforeExpiring); @Override void write(long timestamp, MESSAGE_TYPE message, Offset offset); @Override Path close();  }
@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); }