HeartbeatConsumer implements GarmadonReader.GarmadonMessageHandler { public void run() { for (Map.Entry<Integer, Offset> partitionOffset : latestPartitionsOffset.entrySet()) { int partition = partitionOffset.getKey(); Offset offset = partitionOffset.getValue(); Offset latestHeartbeat = latestHeartbeats.get(partition); if (!offset.equals(latestHeartbeat)) { latestHeartbeats.put(partition, offset); writers.forEach(writer -> writer.heartbeat(partition, offset)); } } } HeartbeatConsumer(Collection<PartitionedWriter<MESSAGE_KIND>> writers, TemporalAmount period); void run(); @Override void handle(GarmadonMessage msg); void dropPartition(int partition);  }
@Test(timeout = 3000) public void writerExpirerStopWhileWaiting() throws InterruptedException { final HeartbeatConsumer hb = new HeartbeatConsumer<String>(Collections.singleton(mock(PartitionedWriter.class)), Duration.ofMillis(10)); hb.run(); }