@Test public void expireNoWriter() { final PartitionedWriter partitionedWriter = new PartitionedWriter<String>(null, new FixedOffsetComputer("ignored", 0), "ignored", DataAccessEventProtos.FsEvent.newBuilder(), mock(Checkpointer.class)); partitionedWriter.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);  }