@Test(timeout = 3000) public void writerExpirer() throws InterruptedException { final PartitionedWriter<String> firstConsumer = mock(PartitionedWriter.class); final PartitionedWriter<String> secondConsumer = mock(PartitionedWriter.class); final PartitionedWriter.Expirer<String> expirer = new PartitionedWriter.Expirer<>( Arrays.asList(firstConsumer, secondConsumer), Duration.ofMillis(1)); expirer.run(); verify(firstConsumer, atLeastOnce()).expireConsumers(); verify(secondConsumer, atLeastOnce()).expireConsumers(); }
void expireConsumers() { possiblyCloseConsumers(ExpiringConsumer::isExpired); }
PartitionedWriter implements Closeable { void expireConsumers() { possiblyCloseConsumers(ExpiringConsumer::isExpired); } }
PartitionedWriter implements Closeable { void expireConsumers() { possiblyCloseConsumers(ExpiringConsumer::isExpired); } PartitionedWriter(BiFunction<Integer, LocalDateTime, ExpiringConsumer<MESSAGE_KIND>> writerBuilder,
                             OffsetComputer offsetComputer, String eventName, Message.Builder emptyMessageBuilder,
                             Checkpointer checkpointer); }
PartitionedWriter implements Closeable { void expireConsumers() { possiblyCloseConsumers(ExpiringConsumer::isExpired); } PartitionedWriter(BiFunction<Integer, LocalDateTime, ExpiringConsumer<MESSAGE_KIND>> writerBuilder,
                             OffsetComputer offsetComputer, String eventName, Message.Builder emptyMessageBuilder,
                             Checkpointer checkpointer); void dropPartition(int partitionId); void write(Instant when, Offset offset, MESSAGE_KIND msg); @Override void close(); Map<Integer, Long> getStartingOffsets(Collection<Integer> partitionsId); void heartbeat(int partition, Offset offset); }
PartitionedWriter implements Closeable { void expireConsumers() { possiblyCloseConsumers(ExpiringConsumer::isExpired); } PartitionedWriter(BiFunction<Integer, LocalDateTime, ExpiringConsumer<MESSAGE_KIND>> writerBuilder,
                             OffsetComputer offsetComputer, String eventName, Message.Builder emptyMessageBuilder,
                             Checkpointer checkpointer); void dropPartition(int partitionId); void write(Instant when, Offset offset, MESSAGE_KIND msg); @Override void close(); Map<Integer, Long> getStartingOffsets(Collection<Integer> partitionsId); void heartbeat(int partition, Offset offset);  }