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