+ partitions.forEach(tp ->
+ {
+ log.info("{} - removing partition: {}", id, tp);
+ PartitionStatistics removed = seen.remove(tp);
+ for (KeyCounter counter : removed.getStatistics())
+ {
+ log.info(
+ "{} - Seen {} messages for partition={}|key={}",
+ id,
+ counter.getResult(),
+ removed.getPartition(),
+ counter.getKey());
+ }
+ repository.save(new StatisticsDocument(removed, consumer.position(tp)));
+ });