ExpiringConsumer implements CloseableWriter<MESSAGE_TYPE> { boolean isExpired() { Instant expirationInstant = startDate.plus(expirationDelay); return messagesReceived >= messagesBeforeExpiring || Instant.now().isAfter(expirationInstant); } 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 expiredFromTheStart() throws InterruptedException { final ExpiringConsumer<Integer> noTimeConsumer = new ExpiringConsumer<>(null, Duration.ZERO, 10); Thread.sleep(1); Assert.assertTrue(noTimeConsumer.isExpired()); final ExpiringConsumer<Integer> noMessageConsumer = new ExpiringConsumer<>(null, Duration.ofMinutes(10), 0); Assert.assertTrue(noMessageConsumer.isExpired()); final ExpiringConsumer<Integer> allZerosConsumer = new ExpiringConsumer<>(null, Duration.ZERO, 0); Thread.sleep(1); Assert.assertTrue(allZerosConsumer.isExpired()); }